diff --git a/docs/content/en/docs/documentation/eventing.md b/docs/content/en/docs/documentation/eventing.md index 4b890407e4..431c1b0eb8 100644 --- a/docs/content/en/docs/documentation/eventing.md +++ b/docs/content/en/docs/documentation/eventing.md @@ -222,6 +222,35 @@ 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)); +``` + +With virtual threads enabled, see [virtual threads](../operations/configuration/#virtual-threads), the +polls run on virtual threads, the executor only keeping one platform thread to do the timing; the +number of threads then caps how many polls run at the same time. + +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) diff --git a/docs/content/en/docs/documentation/operations/configuration.md b/docs/content/en/docs/documentation/operations/configuration.md index 182fd19555..71124dd419 100644 --- a/docs/content/en/docs/documentation/operations/configuration.md +++ b/docs/content/en/docs/documentation/operations/configuration.md @@ -23,6 +23,55 @@ Operator operator = new Operator( override -> override .withLeaderElectionConfiguration(new LeaderElectionConfiguration("bar", "barNS"))); ``` +### Virtual Threads + +Reconciliation is mostly about blocking: talking to the Kubernetes API server or to external +systems. Virtual threads make such blocking calls much cheaper than platform threads, and the +framework can be switched over to them with a single flag: + +```java +Operator operator = new Operator(override -> override.withUseVirtualThreads(true)); +``` + +When enabled, reconciliations, dependent resource workflows and the framework's internal +housekeeping (starting the informers, for example) all run on virtual threads. + +Enabling virtual threads does **not** remove the concurrency limits, parallelism is configured +exactly as before: `withConcurrentReconciliationThreads(int)` still caps how many reconciliations +run at the same time and `withConcurrentWorkflowExecutorThreads(int)` how many dependent resources +of a workflow are processed concurrently. Only the threads backing those limits change. Since +virtual threads are cheap, these limits can usually be raised significantly compared to what is +reasonable with platform threads. + +Things to keep in mind: + +- Virtual threads are only supported on **Java 25 or later**. Before Java 25 a virtual thread pins + its carrier thread while it is inside a `synchronized` block, which can starve the carrier pool + and even deadlock the operator on startup. [JEP 491](https://openjdk.org/jeps/491), delivered in + Java 25, removed that pinning. This is checked at runtime: on an older JVM, including the ones + where virtual threads exist (Java 21 to 24), a warning is logged and platform threads are used + instead, so the same configuration works on any Java version the operator runs on. +- The scheduled tasks (the polling event sources, and the retried and rescheduled reconciliations) + run on virtual threads too. The JDK has no scheduled executor backed by virtual threads, so the + default ones keep a single platform thread for the timing only and run each task they fire on a + virtual thread. `withConcurrentScheduledTaskThreads(int)` and + `withConcurrentRetryAndRescheduleThreads(int)` still cap how many of those tasks run at the same + time. +- With virtual threads, which are always daemon threads, the framework keeps one idle platform + thread (`josdk-keep-alive`) running until the operator is stopped, so that an operator whose + `main` method returns after starting it doesn't let the JVM exit. +- A custom `ExecutorService` provided through `withExecutorService(...)` or + `withWorkflowExecutorService(...)` is always used as is, the flag has no effect on it. +- The Kubernetes client the framework creates when none is provided also switches its internal + task executor (used to dispatch informer events to their handlers and to deliver watch events) + to virtual threads. A client you provide through `withKubernetesClient(...)` is used as is: to + get the same behavior, configure it yourself, e.g. with + `new KubernetesClientBuilder().withTaskExecutor(Executors.newVirtualThreadPerTaskExecutor())` + (the client doesn't shut down an executor passed that way, use `withTaskExecutorSupplier(...)` + if it should be shut down when the client is closed). + Either way, the blocking calls your reconciler makes through the client work well on virtual + threads without any client-side change. + ## Reconciler-Level Configuration While reconcilers are typically configured using the `@ControllerConfiguration` annotation, you can also override configuration at runtime when registering the reconciler with the operator. You can either: @@ -267,6 +316,7 @@ All operator-level keys are prefixed with `josdk.`. |---|---|---| | `josdk.check-crd` | `Boolean` | Validate CRDs against local model on startup | | `josdk.close-client-on-stop` | `Boolean` | Close the Kubernetes client when the operator stops | +| `josdk.use-virtual-threads` | `Boolean` | Run the framework's concurrent work on virtual threads (only supported on Java 25+, platform threads are used on older versions) | | `josdk.use-ssa-to-patch-primary-resource` | `Boolean` | Use Server-Side Apply to patch the primary resource | | `josdk.clone-secondary-resources-when-getting-from-cache` | `Boolean` | Clone secondary resources on cache reads | @@ -283,6 +333,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 | diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ExecutorUtilsInternal.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ExecutorUtilsInternal.java new file mode 100644 index 0000000000..c989a9e0db --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/ExecutorUtilsInternal.java @@ -0,0 +1,69 @@ +/* + * 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; + +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledThreadPoolExecutor; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.atomic.AtomicLong; + +public class ExecutorUtilsInternal { + + // prevent instantiation of util class + private ExecutorUtilsInternal() {} + + /** + * Creates a {@link ThreadFactory} producing daemon threads named after the specified prefix. + * Daemon threads don't keep the JVM alive if the {@link Operator} is never stopped, and naming + * them makes the pool they belong to identifiable in thread dumps. + * + * @param namePrefix the prefix the created threads are named after + * @return a {@link ThreadFactory} creating named daemon threads + */ + public static ThreadFactory daemonThreadFactory(String namePrefix) { + final var defaultThreadFactory = Executors.defaultThreadFactory(); + final var counter = new AtomicLong(); + return runnable -> { + final var thread = defaultThreadFactory.newThread(runnable); + thread.setName(namePrefix + "-" + counter.incrementAndGet()); + thread.setDaemon(true); + return thread; + }; + } + + /** + * Creates the kind of {@link ScheduledExecutorService} the operator runs its scheduled (i.e. + * periodic or delayed) tasks on: one whose threads are daemon threads named after the specified + * prefix, and which doesn't let the tasks that were scheduled for later delay its shutdown. + * + * @param corePoolSize the number of threads to keep in the pool + * @param threadNamePrefix the prefix the threads of the pool are named after + * @return a {@link ScheduledExecutorService} to run scheduled tasks on + */ + public static ScheduledExecutorService daemonScheduledThreadPool( + int corePoolSize, String threadNamePrefix) { + final var executor = + new ScheduledThreadPoolExecutor(corePoolSize, daemonThreadFactory(threadNamePrefix)); + // tasks that are scheduled far out (a reconciliation rescheduled in an hour, say) would + // otherwise keep the pool from terminating until the graceful shutdown timeout expires + executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false); + // cancelled tasks are frequent (every retry that is superseded by a new event cancels one) and + // would otherwise be retained until their delay elapses + executor.setRemoveOnCancelPolicy(true); + return executor; + } +} 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 336af9e2a8..e729744e8d 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 @@ -20,7 +20,7 @@ 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; @@ -64,6 +64,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 @@ -155,6 +169,12 @@ public R clone(R object) { * io.javaoperatorsdk.operator.Operator#Operator(Consumer)}, passing your custom instance with * {@link ConfigurationServiceOverrider#withKubernetesClient(KubernetesClient)}. * + *

When {@link #useVirtualThreads()} is enabled (and supported by the JVM), the default client + * runs its internal asynchronous tasks, such as dispatching informer events to their handlers and + * delivering watch events, on virtual threads instead of its default cached platform thread pool. + * A client provided by overriding this method or through {@link + * ConfigurationServiceOverrider#withKubernetesClient(KubernetesClient)} is used as is. + * *

NOTE: It is strongly suggested that implementors override this method since the * default implementation creates a new {@link KubernetesClient} instance each time this method is * called. @@ -163,13 +183,17 @@ public R clone(R object) { * @since 4.4.0 */ default KubernetesClient getKubernetesClient() { - return new KubernetesClientBuilder() - .withConfig( - new ConfigBuilder(Config.autoConfigure(null)) - .withMaxConcurrentRequests(DEFAULT_MAX_CONCURRENT_REQUEST) - .build()) - .withKubernetesSerialization(new KubernetesSerialization()) - .build(); + final var builder = + new KubernetesClientBuilder() + .withConfig( + new ConfigBuilder(Config.autoConfigure(null)) + .withMaxConcurrentRequests(DEFAULT_MAX_CONCURRENT_REQUEST) + .build()) + .withKubernetesSerialization(new KubernetesSerialization()); + if (VirtualThreads.shouldUse(useVirtualThreads())) { + builder.withTaskExecutorSupplier(VirtualThreads.newKubernetesClientTaskExecutorSupplier()); + } + return builder.build(); } /** @@ -203,6 +227,11 @@ default boolean checkCRDAndValidateLocalModel() { * The number of threads the operator can spin out to dispatch reconciliation requests to * reconcilers with the default executors * + *

This is a concurrency limit and applies regardless of whether the default executor is backed + * by platform or by virtual threads, see {@link #useVirtualThreads()}: with virtual threads it + * caps how many reconciliations run at the same time rather than the size of a thread pool. Since + * virtual threads are cheap, the limit can be set considerably higher when they are enabled. + * * @return the number of concurrent reconciliation threads */ default int concurrentReconciliationThreads() { @@ -213,12 +242,45 @@ default int concurrentReconciliationThreads() { * Number of threads the operator can spin out to be used in the workflows with the default * executor. * + *

This is a concurrency limit and applies regardless of whether the default executor is backed + * by platform or by virtual threads, see {@link #useVirtualThreads()}: with virtual threads it + * caps how many dependent resources are processed at the same time rather than the size of a + * thread pool. Since virtual threads are cheap, the limit can be set considerably higher when + * they are enabled. + * * @return the maximum number of concurrent workflow threads */ 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 * @@ -228,6 +290,45 @@ default Metrics getMetrics() { return Metrics.NOOP; } + /** + * Whether the framework should run the tasks it executes concurrently — reconciliations, + * dependent workflows and internal housekeeping such as starting the informers — on virtual + * threads instead of platform threads. + * + *

Virtual threads make blocking operations, which is essentially all a reconciler does while + * talking to the Kubernetes API server or to external systems, much cheaper. Enabling them does + * not lift the configured concurrency limits: {@link #concurrentReconciliationThreads()} + * and {@link #concurrentWorkflowExecutorThreads()} still cap how many reconciliations, + * respectively dependent resources, are processed at the same time, they just aren't backed by a + * pool of platform threads anymore. Since virtual threads are cheap, those limits can be set + * considerably higher than what would be reasonable for platform threads. + * + *

Virtual threads are only supported on Java 25 or later, where JEP 491 removed the pinning of the carrier thread by + * {@code synchronized} blocks. This is checked at runtime: on an older JVM, a warning is logged + * and platform threads are used, so that the same configuration works regardless of the Java + * version the operator runs on. + * + *

Note that this only affects the executors created by the framework: a custom {@link + * ExecutorService} provided through {@link #getExecutorService()} or {@link + * #getWorkflowExecutorService()}, or a custom {@link ScheduledExecutorService} provided through + * {@link #getScheduledExecutorService()} or {@link #getRetryAndRescheduleExecutorService()}, is + * used as is. The default scheduled executors keep a single platform thread for the timing and + * run each task they fire on a virtual thread. The same goes for the {@link KubernetesClient}: + * the default client created by {@link #getKubernetesClient()} then also runs its internal + * asynchronous tasks (informer event dispatching, watch event delivery) on virtual threads, while + * a custom client, e.g. provided through {@link + * ConfigurationServiceOverrider#withKubernetesClient(KubernetesClient)}, keeps whatever task + * executor it was built with. The blocking calls a reconciler performs through the client don't + * need any of this: they already park the calling virtual thread rather than its carrier. + * + * @return {@code true} to use virtual threads, {@code false} (default) to use platform threads + * @since 5.7.0 + */ + default boolean useVirtualThreads() { + return false; + } + /** * Override to provide a custom {@link ExecutorService} implementation to change how threads * handle concurrent reconciliations @@ -236,7 +337,8 @@ default Metrics getMetrics() { * processing */ default ExecutorService getExecutorService() { - return Executors.newFixedThreadPool(concurrentReconciliationThreads()); + return ExecutorServiceManager.newBoundedExecutorService( + concurrentReconciliationThreads(), useVirtualThreads()); } /** @@ -246,7 +348,46 @@ default ExecutorService getExecutorService() { * @return the {@link ExecutorService} implementation to use for dependent workflow processing */ default ExecutorService getWorkflowExecutorService() { - return Executors.newFixedThreadPool(concurrentWorkflowExecutorThreads()); + return ExecutorServiceManager.newBoundedExecutorService( + concurrentWorkflowExecutorThreads(), useVirtualThreads()); + } + + /** + * 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. + * + *

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 ExecutorServiceManager.newScheduledExecutorService( + concurrentScheduledTaskThreads(), useVirtualThreads(), "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. + * + *

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 ExecutorServiceManager.newScheduledExecutorService( + concurrentRetryAndRescheduleThreads(), useVirtualThreads(), "josdk-retry-reschedule"); } /** 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 2cf6540af0..7215cc29ba 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 @@ -21,6 +21,7 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutorService; +import java.util.concurrent.ScheduledExecutorService; import java.util.function.Function; import org.slf4j.Logger; @@ -45,11 +46,16 @@ public class ConfigurationServiceOverrider { private Boolean checkCR; private Integer concurrentReconciliationThreads; private Integer concurrentWorkflowExecutorThreads; + private Integer concurrentScheduledTaskThreads; + private Integer concurrentRetryAndRescheduleThreads; private Cloner cloner; private Boolean closeClientOnStop; private KubernetesClient client; private ExecutorService executorService; private ExecutorService workflowExecutorService; + private Boolean useVirtualThreads; + private ScheduledExecutorService scheduledExecutorService; + private ScheduledExecutorService retryAndRescheduleExecutorService; private LeaderElectionConfiguration leaderElectionConfiguration; private String clusterScopedEventNamespace; private EventRecorder eventRecorder; @@ -86,6 +92,34 @@ public ConfigurationServiceOverrider withConcurrentWorkflowExecutorThreads(int t return this; } + /** + * Sets the number of threads used to run the operator's scheduled (i.e. periodic or delayed) + * tasks, which are shared by all its polling event sources. + * + * @param threadNumber the maximum number of concurrent scheduled task threads + * @return this {@link ConfigurationServiceOverrider} for chained customization + * @see ConfigurationService#concurrentScheduledTaskThreads() + * @since 5.6.0 + */ + public ConfigurationServiceOverrider withConcurrentScheduledTaskThreads(int threadNumber) { + this.concurrentScheduledTaskThreads = threadNumber; + return this; + } + + /** + * Sets the number of threads used to trigger the operator's retried and rescheduled + * reconciliations, which are shared by all its controllers. + * + * @param threadNumber the maximum number of concurrent retry and reschedule threads + * @return this {@link ConfigurationServiceOverrider} for chained customization + * @see ConfigurationService#concurrentRetryAndRescheduleThreads() + * @since 5.6.0 + */ + public ConfigurationServiceOverrider withConcurrentRetryAndRescheduleThreads(int threadNumber) { + this.concurrentRetryAndRescheduleThreads = threadNumber; + return this; + } + @SuppressWarnings("rawtypes") public ConfigurationServiceOverrider withDependentResourceFactory( DependentResourceFactory dependentResourceFactory) { @@ -119,6 +153,50 @@ public ConfigurationServiceOverrider withWorkflowExecutorService( return this; } + /** + * Replaces the executor used to run the operator's scheduled (i.e. periodic or delayed) tasks, + * which are shared by all its polling event sources. + * + * @param scheduledExecutorService the executor to run scheduled tasks on + * @return this {@link ConfigurationServiceOverrider} for chained customization + * @see ConfigurationService#getScheduledExecutorService() + * @since 5.6.0 + */ + public ConfigurationServiceOverrider withScheduledExecutorService( + ScheduledExecutorService scheduledExecutorService) { + this.scheduledExecutorService = scheduledExecutorService; + return this; + } + + /** + * Replaces the executor used to trigger the operator's retried and rescheduled reconciliations, + * which is shared by all its controllers. + * + * @param retryAndRescheduleExecutorService the executor to trigger retried and rescheduled + * reconciliations on + * @return this {@link ConfigurationServiceOverrider} for chained customization + * @see ConfigurationService#getRetryAndRescheduleExecutorService() + * @since 5.6.0 + */ + public ConfigurationServiceOverrider withRetryAndRescheduleExecutorService( + ScheduledExecutorService retryAndRescheduleExecutorService) { + this.retryAndRescheduleExecutorService = retryAndRescheduleExecutorService; + return this; + } + + /** + * Makes the framework run the tasks it executes concurrently on virtual threads instead of + * platform threads. Only supported on Java 25 or later, see {@link + * ConfigurationService#useVirtualThreads()} for the details. + * + * @param useVirtualThreads {@code true} to use virtual threads + * @return this {@link ConfigurationServiceOverrider} for chained customization + */ + public ConfigurationServiceOverrider withUseVirtualThreads(boolean useVirtualThreads) { + this.useVirtualThreads = useVirtualThreads; + return this; + } + /** * Replaces the default {@link KubernetesClient} instance by the specified one. This is the * preferred mechanism to configure which client will be used to access the cluster. @@ -312,6 +390,28 @@ public int concurrentWorkflowExecutorThreads() { original.concurrentWorkflowExecutorThreads()); } + @Override + public int concurrentScheduledTaskThreads() { + return Utils.ensureValid( + overriddenValueOrDefault( + concurrentScheduledTaskThreads, + ConfigurationService::concurrentScheduledTaskThreads), + "maximum scheduled task threads", + 1, + original.concurrentScheduledTaskThreads()); + } + + @Override + public int concurrentRetryAndRescheduleThreads() { + return Utils.ensureValid( + overriddenValueOrDefault( + concurrentRetryAndRescheduleThreads, + ConfigurationService::concurrentRetryAndRescheduleThreads), + "maximum retry and reschedule threads", + 1, + original.concurrentRetryAndRescheduleThreads()); + } + @Override public Metrics getMetrics() { return overriddenValueOrDefault(metrics, ConfigurationService::getMetrics); @@ -322,6 +422,11 @@ public boolean closeClientOnStop() { return overriddenValueOrDefault(closeClientOnStop, ConfigurationService::closeClientOnStop); } + @Override + public boolean useVirtualThreads() { + return overriddenValueOrDefault(useVirtualThreads, ConfigurationService::useVirtualThreads); + } + @Override public ExecutorService getExecutorService() { if (executorService != null) { @@ -340,6 +445,24 @@ public ExecutorService getWorkflowExecutorService() { } } + @Override + public ScheduledExecutorService getScheduledExecutorService() { + if (scheduledExecutorService != null) { + return scheduledExecutorService; + } else { + return super.getScheduledExecutorService(); + } + } + + @Override + public ScheduledExecutorService getRetryAndRescheduleExecutorService() { + if (retryAndRescheduleExecutorService != null) { + return retryAndRescheduleExecutorService; + } else { + return super.getRetryAndRescheduleExecutorService(); + } + } + @Override public Optional getLeaderElectionConfiguration() { return leaderElectionConfiguration != null diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java index cdcafcaa46..2ca1591536 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManager.java @@ -26,6 +26,7 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.LockSupport; import java.util.function.Function; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -33,6 +34,7 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import io.javaoperatorsdk.operator.ExecutorUtilsInternal; import io.javaoperatorsdk.operator.OperatorException; public class ExecutorServiceManager { @@ -42,13 +44,66 @@ public class ExecutorServiceManager { private ExecutorService workflowExecutor; private ExecutorService cachingExecutorService; private ScheduledExecutorService scheduledExecutorService; + private ScheduledExecutorService retryAndRescheduleExecutorService; private boolean started; private ConfigurationService configurationService; + private Thread keepAliveThread; ExecutorServiceManager(ConfigurationService configurationService) { start(configurationService); } + /** + * Creates the executor service used to run a bounded number of tasks concurrently, either backed + * by virtual threads or by a fixed size pool of platform threads. The concurrency limit is + * enforced in both cases. + * + * @param maxConcurrency the maximal number of tasks executed at the same time + * @param useVirtualThreads whether virtual threads should be used, see {@link + * ConfigurationService#useVirtualThreads()} + * @return the created {@link ExecutorService} + */ + public static ExecutorService newBoundedExecutorService( + int maxConcurrency, boolean useVirtualThreads) { + return VirtualThreads.shouldUse(useVirtualThreads) + ? VirtualThreads.newBoundedVirtualThreadExecutor(maxConcurrency) + : Executors.newFixedThreadPool(maxConcurrency); + } + + /** + * Creates the executor service used to run scheduled (i.e. periodic or delayed) tasks, either + * firing them on virtual threads or on a pool of platform threads. The platform thread variant is + * the one of {@link ExecutorUtilsInternal#daemonScheduledThreadPool(int, String)}. In both cases + * the threads are daemon threads and the concurrency limit is enforced. + * + * @param maxConcurrency the maximal number of tasks executed at the same time + * @param useVirtualThreads whether virtual threads should be used, see {@link + * ConfigurationService#useVirtualThreads()} + * @param threadNamePrefix the prefix the threads of the executor are named after + * @return the {@link ScheduledExecutorService} to run scheduled tasks on + * @since 5.7.0 + */ + public static ScheduledExecutorService newScheduledExecutorService( + int maxConcurrency, boolean useVirtualThreads, String threadNamePrefix) { + return VirtualThreads.shouldUse(useVirtualThreads) + ? VirtualThreads.newScheduledExecutor(maxConcurrency, threadNamePrefix) + : ExecutorUtilsInternal.daemonScheduledThreadPool(maxConcurrency, threadNamePrefix); + } + + /** + * Creates the executor service used to run an unbounded number of tasks concurrently, either + * backed by virtual threads or by a cached pool of platform threads. + * + * @param useVirtualThreads whether virtual threads should be used, see {@link + * ConfigurationService#useVirtualThreads()} + * @return the created {@link ExecutorService} + */ + public static ExecutorService newUnboundedExecutorService(boolean useVirtualThreads) { + return VirtualThreads.shouldUse(useVirtualThreads) + ? VirtualThreads.newVirtualThreadPerTaskExecutor() + : Executors.newCachedThreadPool(); + } + /** * Uses cachingExecutorService from this manager. Use this only for tasks, that don't have dynamic * nature, in sense that won't grow with the number of inputs (thus kubernetes resources) @@ -128,30 +183,81 @@ public ExecutorService cachingExecutorService() { return cachingExecutorService; } + /** + * The executor the operator runs its scheduled (i.e. periodic or delayed) tasks on, shared by its + * polling event sources. Note that it is only valid while the manager is started: it is shut down + * by {@link #stop(Duration)} and replaced by a fresh one on the next {@link + * #start(ConfigurationService)}, so callers should retrieve it when they start rather than hold + * on to it. + * + * @return the executor to run scheduled tasks on + */ public ScheduledExecutorService scheduledExecutorService() { return scheduledExecutorService; } + /** + * The executor the operator triggers its retried and rescheduled reconciliations on, kept + * separate from {@link #scheduledExecutorService()} so that a slow poll can't delay a retry. The + * same lifecycle caveat as for {@link #scheduledExecutorService()} applies. + * + * @return the executor to trigger retried and rescheduled reconciliations on + */ + public ScheduledExecutorService retryAndRescheduleExecutorService() { + return retryAndRescheduleExecutorService; + } + public synchronized void start(ConfigurationService configurationService) { if (!started) { this.configurationService = configurationService; // used to lazy init workflow executor - this.cachingExecutorService = Executors.newCachedThreadPool(); - this.scheduledExecutorService = Executors.newScheduledThreadPool(0); + this.cachingExecutorService = + newUnboundedExecutorService(configurationService.useVirtualThreads()); + this.scheduledExecutorService = configurationService.getScheduledExecutorService(); + this.retryAndRescheduleExecutorService = + configurationService.getRetryAndRescheduleExecutorService(); this.executor = new InstrumentedExecutorService(configurationService.getExecutorService()); + if (VirtualThreads.shouldUse(configurationService.useVirtualThreads())) { + startKeepAliveThread(); + } started = true; } } + /** + * Virtual threads are always daemon threads, so with them nothing is left to keep the JVM running + * once the main method of an operator returns after starting it, as opposed to the non-daemon + * threads of the platform thread pools. This thread restores that, until the executors are + * stopped. + */ + private void startKeepAliveThread() { + keepAliveThread = + new Thread( + () -> { + while (!Thread.currentThread().isInterrupted()) { + LockSupport.park(); + } + }, + "josdk-keep-alive"); + keepAliveThread.setDaemon(false); + keepAliveThread.start(); + } + public synchronized void stop(Duration gracefulShutdownTimeout) { - var parallelExec = Executors.newFixedThreadPool(4); + if (keepAliveThread != null) { + keepAliveThread.interrupt(); + keepAliveThread = null; + } + var shutdowns = + List.of( + shutdown(executor, gracefulShutdownTimeout), + shutdown(workflowExecutor, gracefulShutdownTimeout), + shutdown(cachingExecutorService, gracefulShutdownTimeout), + shutdown(scheduledExecutorService, gracefulShutdownTimeout), + shutdown(retryAndRescheduleExecutorService, gracefulShutdownTimeout)); + var parallelExec = Executors.newFixedThreadPool(shutdowns.size()); try { log.debug("Closing executor"); - parallelExec.invokeAll( - List.of( - shutdown(executor, gracefulShutdownTimeout), - shutdown(workflowExecutor, gracefulShutdownTimeout), - shutdown(cachingExecutorService, gracefulShutdownTimeout), - shutdown(scheduledExecutorService, gracefulShutdownTimeout))); + parallelExec.invokeAll(shutdowns); } catch (InterruptedException e) { log.debug("Exception closing executor: {}", e.getLocalizedMessage()); Thread.currentThread().interrupt(); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreadScheduledExecutor.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreadScheduledExecutor.java new file mode 100644 index 0000000000..d38c5abe96 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreadScheduledExecutor.java @@ -0,0 +1,305 @@ +/* + * 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.api.config; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.Callable; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Delayed; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import io.javaoperatorsdk.operator.ExecutorUtilsInternal; + +/** + * A {@link ScheduledExecutorService} running the tasks it fires on the threads of another executor, + * which is how scheduled tasks end up on virtual threads: the JDK has no virtual thread backed + * {@link ScheduledExecutorService}, so a single daemon platform thread does the timing and only + * hands each fired task off. + * + *

The semantics follow the ones of the default scheduled executor of the SDK, see {@link + * ExecutorUtilsInternal#daemonScheduledThreadPool(int, String)}: tasks that were scheduled for + * later, and periodic tasks, are discarded when the executor is shut down, while tasks that already + * started are let finish. + */ +final class VirtualThreadScheduledExecutor extends AbstractExecutorService + implements ScheduledExecutorService { + + private final ExecutorService worker; + private final ScheduledExecutorService timer; + private final Set> waiting = ConcurrentHashMap.newKeySet(); + private final Object lifecycleLock = new Object(); + + VirtualThreadScheduledExecutor(ExecutorService worker, String threadNamePrefix) { + this.worker = worker; + this.timer = ExecutorUtilsInternal.daemonScheduledThreadPool(1, threadNamePrefix + "-timer"); + } + + @Override + public ScheduledFuture schedule(Runnable command, long delay, TimeUnit unit) { + Objects.requireNonNull(command); + return start(new Handle(Executors.callable(command, null), 0), delay, unit); + } + + @Override + public ScheduledFuture schedule(Callable callable, long delay, TimeUnit unit) { + Objects.requireNonNull(callable); + return start(new Handle<>(callable, 0), delay, unit); + } + + @Override + public ScheduledFuture scheduleAtFixedRate( + Runnable command, long initialDelay, long period, TimeUnit unit) { + Objects.requireNonNull(command); + if (period <= 0) { + throw new IllegalArgumentException("period must be positive"); + } + return start( + new Handle(Executors.callable(command, null), unit.toNanos(period)), + initialDelay, + unit); + } + + @Override + public ScheduledFuture scheduleWithFixedDelay( + Runnable command, long initialDelay, long delay, TimeUnit unit) { + Objects.requireNonNull(command); + if (delay <= 0) { + throw new IllegalArgumentException("delay must be positive"); + } + return start( + new Handle(Executors.callable(command, null), -unit.toNanos(delay)), + initialDelay, + unit); + } + + private Handle start(Handle handle, long initialDelay, TimeUnit unit) { + handle.arm(Math.max(0, unit.toNanos(initialDelay))); + return handle; + } + + @Override + public void execute(Runnable command) { + worker.execute(Objects.requireNonNull(command)); + } + + @Override + public void shutdown() { + synchronized (lifecycleLock) { + cancelWaiting(false); + timer.shutdown(); + worker.shutdown(); + } + } + + @Override + public List shutdownNow() { + final var notStarted = new ArrayList(); + synchronized (lifecycleLock) { + cancelWaiting(true); + timer.shutdownNow(); + for (var task : worker.shutdownNow()) { + if (task instanceof Dispatched dispatched) { + dispatched.handle.cancel(false); + } else { + notStarted.add(task); + } + } + } + return Collections.unmodifiableList(notStarted); + } + + private void cancelWaiting(boolean mayInterruptIfRunning) { + for (var handle : waiting) { + handle.cancel(mayInterruptIfRunning); + } + } + + @Override + public boolean isShutdown() { + return timer.isShutdown() && worker.isShutdown(); + } + + @Override + public boolean isTerminated() { + return timer.isTerminated() && worker.isTerminated(); + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { + final var deadline = System.nanoTime() + unit.toNanos(timeout); + return timer.awaitTermination(timeout, unit) + && worker.awaitTermination(deadline - System.nanoTime(), TimeUnit.NANOSECONDS); + } + + /** + * What is handed to the worker once a task fires, so that it can be traced back to its handle. + */ + private record Dispatched(Handle handle) implements Runnable { + @Override + public void run() { + handle.runOnWorker(); + } + } + + /** + * The future of a scheduled task. A positive {@code period} makes it a task at a fixed rate, a + * negative one a task with a fixed delay, whose absolute value is the delay. + */ + private final class Handle implements ScheduledFuture { + + private final Callable callable; + private final long period; + private final CompletableFuture result = new CompletableFuture<>(); + + private volatile ScheduledFuture timerFuture; + private volatile Thread runner; + private volatile long triggerTime; + + private Handle(Callable callable, long period) { + this.callable = callable; + this.period = period; + } + + private void arm(long delayNanos) { + triggerTime = System.nanoTime() + delayNanos; + synchronized (lifecycleLock) { + waiting.add(this); + try { + timerFuture = timer.schedule(this::dispatch, delayNanos, TimeUnit.NANOSECONDS); + } catch (RejectedExecutionException e) { + waiting.remove(this); + throw e; + } + } + if (result.isDone()) { + timerFuture.cancel(false); + } + } + + private void dispatch() { + waiting.remove(this); + if (result.isDone()) { + return; + } + try { + worker.execute(new Dispatched(this)); + } catch (RejectedExecutionException e) { + result.cancel(false); + } + } + + private void runOnWorker() { + if (result.isDone()) { + return; + } + runner = Thread.currentThread(); + try { + if (period == 0) { + result.complete(callable.call()); + return; + } + callable.call(); + } catch (Throwable t) { + result.completeExceptionally(t); + return; + } finally { + runner = null; + } + rearm(); + } + + private void rearm() { + if (result.isDone()) { + return; + } + final long delay; + if (period > 0) { + delay = Math.max(0, triggerTime + period - System.nanoTime()); + } else { + delay = -period; + } + try { + arm(delay); + } catch (RejectedExecutionException e) { + result.cancel(false); + } + } + + @Override + public boolean cancel(boolean mayInterruptIfRunning) { + final var cancelled = result.cancel(mayInterruptIfRunning); + if (cancelled) { + waiting.remove(this); + final var future = timerFuture; + if (future != null) { + future.cancel(false); + } + final var thread = runner; + if (mayInterruptIfRunning && thread != null) { + thread.interrupt(); + } + } + return cancelled; + } + + @Override + public boolean isCancelled() { + return result.isCancelled(); + } + + @Override + public boolean isDone() { + return result.isDone(); + } + + @Override + public V get() throws InterruptedException, ExecutionException { + return result.get(); + } + + @Override + public V get(long timeout, TimeUnit unit) + throws InterruptedException, ExecutionException, TimeoutException { + return result.get(timeout, unit); + } + + @Override + public long getDelay(TimeUnit unit) { + return unit.convert(triggerTime - System.nanoTime(), TimeUnit.NANOSECONDS); + } + + @Override + public int compareTo(Delayed other) { + if (other == this) { + return 0; + } + return Long.compare(getDelay(TimeUnit.NANOSECONDS), other.getDelay(TimeUnit.NANOSECONDS)); + } + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java new file mode 100644 index 0000000000..0cf512587f --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java @@ -0,0 +1,282 @@ +/* + * 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.api.config; + +import java.lang.invoke.MethodHandle; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.MethodType; +import java.util.ArrayList; +import java.util.List; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.fabric8.kubernetes.client.KubernetesClientBuilder.ExecutorSupplier; +import io.javaoperatorsdk.operator.OperatorException; + +/** + * Creates the virtual thread based executors used when {@link + * ConfigurationService#useVirtualThreads()} is enabled. + * + *

Virtual threads are only supported on Java 25 or later, where JEP 491 removed the pinning of the carrier thread by + * {@code synchronized} blocks. On older JVMs, including the ones where virtual threads already + * exist, the request is ignored with a warning and platform threads are used, see {@link + * #shouldUse(boolean)}. + * + *

The SDK is compiled for Java 17, in which virtual threads don't exist yet, so {@code + * Executors.newVirtualThreadPerTaskExecutor()} is looked up reflectively. + * + *

The JDK provides no virtual thread backed {@link + * java.util.concurrent.ScheduledExecutorService} (the Loom runtime itself uses a platform thread + * scheduler to unpark timed out virtual threads), so the scheduled variant, see {@link + * #newScheduledExecutor(int, String)}, keeps a single platform thread for the timing only and hands + * each fired task off to a virtual thread. + */ +final class VirtualThreads { + + private static final Logger log = LoggerFactory.getLogger(VirtualThreads.class); + + /** The first Java version virtual threads are supported on. */ + static final int MINIMUM_SUPPORTED_JAVA_VERSION = 25; + + private static final MethodHandle NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR = lookupFactoryMethod(); + private static final AtomicBoolean UNSUPPORTED_WARNING_LOGGED = new AtomicBoolean(); + + private VirtualThreads() {} + + private static MethodHandle lookupFactoryMethod() { + try { + return MethodHandles.publicLookup() + .findStatic( + Executors.class, + "newVirtualThreadPerTaskExecutor", + MethodType.methodType(ExecutorService.class)); + } catch (NoSuchMethodException | IllegalAccessException e) { + log.debug("Virtual threads are not available on this JVM", e); + return null; + } + } + + /** + * Whether the JVM the operator runs on supports virtual threads, i.e. is Java {@value + * #MINIMUM_SUPPORTED_JAVA_VERSION} or later. + */ + static boolean isSupported() { + return isSupported(Runtime.version().feature()); + } + + static boolean isSupported(int javaFeatureVersion) { + return NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR != null + && javaFeatureVersion >= MINIMUM_SUPPORTED_JAVA_VERSION; + } + + /** + * Whether virtual threads should effectively be used, i.e. they were requested through {@link + * ConfigurationService#useVirtualThreads()} and the JVM supports them. Requesting them + * on a JVM that doesn't support them is only warned about, so that the same configuration can be + * used regardless of the Java version the operator ends up running on, the only consequence being + * that platform threads are used instead. Concurrency limits are enforced either way. + */ + static boolean shouldUse(boolean requested) { + if (!requested || isSupported()) { + return requested; + } + if (UNSUPPORTED_WARNING_LOGGED.compareAndSet(false, true)) { + log.warn( + "Virtual threads were requested but are only supported on Java {} or later, and the JVM" + + " in use is Java {}. Falling back to platform threads.", + MINIMUM_SUPPORTED_JAVA_VERSION, + Runtime.version().feature()); + } + return false; + } + + /** An unbounded executor starting a new virtual thread for each submitted task. */ + static ExecutorService newVirtualThreadPerTaskExecutor() { + if (!isSupported()) { + throw new OperatorException( + "Virtual threads are only supported on Java " + + MINIMUM_SUPPORTED_JAVA_VERSION + + " or later"); + } + try { + return (ExecutorService) NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR.invokeExact(); + } catch (Throwable e) { + throw new OperatorException("Couldn't create a virtual thread per task executor", e); + } + } + + /** + * Supplies the task executor of the default {@link io.fabric8.kubernetes.client.KubernetesClient} + * created by {@link ConfigurationService#getKubernetesClient()}: an unbounded virtual thread + * executor, replacing the client's default cached platform thread pool, that is shut down when + * the client is closed. + */ + static ExecutorSupplier newKubernetesClientTaskExecutorSupplier() { + return new ExecutorSupplier() { + @Override + public Executor get() { + return newVirtualThreadPerTaskExecutor(); + } + + @Override + public void onClose(Executor executor) { + ((ExecutorService) executor).shutdownNow(); + } + }; + } + + /** + * A {@link ScheduledExecutorService} running the tasks it fires on virtual threads, at most + * {@code maxConcurrency} of them at the same time. The timing is done by a single daemon platform + * thread named after {@code threadNamePrefix}, which only hands the tasks off. + */ + static ScheduledExecutorService newScheduledExecutor( + int maxConcurrency, String threadNamePrefix) { + return new VirtualThreadScheduledExecutor( + newBoundedVirtualThreadExecutor(maxConcurrency), threadNamePrefix); + } + + /** + * A virtual thread based executor executing at most {@code maxConcurrency} tasks at the same + * time, the equivalent of a fixed size platform thread pool. + */ + static ExecutorService newBoundedVirtualThreadExecutor(int maxConcurrency) { + if (maxConcurrency <= 0) { + throw new IllegalArgumentException("maxConcurrency must be positive: " + maxConcurrency); + } + return new BoundedExecutorService(newVirtualThreadPerTaskExecutor(), maxConcurrency); + } + + /** + * Limits how many of the tasks submitted to the wrapped executor run at the same time. + * + *

A thread is started for each task as soon as it is submitted, the task then waits for a + * permit before it actually runs. This only makes sense with virtual threads, which are cheap + * enough to be parked in large numbers, and has the property that submitting a task never blocks + * the submitting thread, just like queuing it on a fixed size platform thread pool wouldn't. + */ + private static final class BoundedExecutorService extends AbstractExecutorService { + + private final ExecutorService delegate; + private final Semaphore permits; + private final Set pending = ConcurrentHashMap.newKeySet(); + private final Object lifecycleLock = new Object(); + + private BoundedExecutorService(ExecutorService delegate, int maxConcurrency) { + this.delegate = delegate; + this.permits = new Semaphore(maxConcurrency, true); + } + + @Override + public void execute(Runnable command) { + Objects.requireNonNull(command); + var task = new PendingTask(command); + // admission is closed under the same lock that shutdownNow drains pending tasks with, so a + // task is either rejected or seen by the drain, and never interrupted while unaccounted for + synchronized (lifecycleLock) { + pending.add(task); + try { + delegate.execute(task); + } catch (RuntimeException e) { + pending.remove(task); + throw e; + } + } + } + + private final class PendingTask implements Runnable { + + private final Runnable command; + private final AtomicBoolean claimed = new AtomicBoolean(); + + private PendingTask(Runnable command) { + this.command = command; + } + + private boolean claim() { + return claimed.compareAndSet(false, true); + } + + @Override + public void run() { + try { + permits.acquire(); + } catch (InterruptedException e) { + // shutdownNow interrupted us and already handed the task back to its caller + Thread.currentThread().interrupt(); + return; + } + try { + if (!claim()) { + return; + } + pending.remove(this); + command.run(); + } finally { + permits.release(); + } + } + } + + @Override + public void shutdown() { + delegate.shutdown(); + } + + @Override + public List shutdownNow() { + var notStarted = new ArrayList(); + synchronized (lifecycleLock) { + for (var task : pending) { + if (task.claim()) { + pending.remove(task); + notStarted.add(task.command); + } + } + delegate.shutdownNow(); + } + return notStarted; + } + + @Override + public boolean isShutdown() { + return delegate.isShutdown(); + } + + @Override + public boolean isTerminated() { + return delegate.isTerminated(); + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { + return delegate.awaitTermination(timeout, unit); + } + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java index 50e61526c4..c5e3e9c4e8 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/DefaultContext.java @@ -50,6 +50,8 @@ public class DefaultContext

implements Context

{ private final boolean primaryResourceDeleted; private final boolean primaryResourceFinalStateUnknown; private final Map, Object> desiredStates = new ConcurrentHashMap<>(); + private final Map, DesiredStateInProgress> desiredStatesInProgress = + new ConcurrentHashMap<>(); private final ResourceOperations

resourceOperations; public DefaultContext( @@ -254,15 +256,40 @@ public DefaultContext

setRetryInfo(RetryInfo retryInfo) { return this; } - @SuppressWarnings("unchecked") public R getOrComputeDesiredStateFor( DependentResource dependentResource, Function desiredStateComputer) { + return getOrComputeDesiredStateFor(dependentResource, dependentResource, desiredStateComputer); + } + + /** + * Same as {@link #getOrComputeDesiredStateFor(DependentResource, Function)} but hands {@code + * aspectTarget} rather than {@code dependentResource} to the {@link DesiredStateAspect}s, for + * dependents that are only a stand-in for the one the user configured. + */ + @SuppressWarnings("unchecked") + public R getOrComputeDesiredStateFor( + DependentResource dependentResource, + DependentResource aspectTarget, + Function desiredStateComputer) { + // an aspect can look up the actual resource of its dependent, which asks for the desired state + // being computed: the thread computing it is handed the state as it is so far, instead of + // re-entering the computation + final var computing = desiredStatesInProgress.get(dependentResource); + if (computing != null && computing.thread() == Thread.currentThread()) { + return (R) computing.desired(); + } return (R) desiredStates.computeIfAbsent( dependentResource, ignored -> { final var desired = desiredStateComputer.apply(getPrimaryResource()); - applyDesiredStateAspects(desired, dependentResource); + desiredStatesInProgress.put( + dependentResource, new DesiredStateInProgress(Thread.currentThread(), desired)); + try { + applyDesiredStateAspects(desired, aspectTarget); + } finally { + desiredStatesInProgress.remove(dependentResource); + } return desired; }); } @@ -273,11 +300,13 @@ public R getOrComputeDesiredStateFor( * resources are therefore left untouched. */ private void applyDesiredStateAspects(Object desired, DependentResource dependentResource) { - if (desired instanceof HasMetadata hasMetadata) { + if (desired instanceof HasMetadata hasMetadata && controllerConfiguration != null) { controllerConfiguration .getConfigurationService() .desiredStateAspects() .forEach(aspect -> aspect.apply(hasMetadata, dependentResource, this)); } } + + private record DesiredStateInProgress(Thread thread, Object desired) {} } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java index 2428eeba96..bfe46332c9 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/BulkDependentResourceReconciler.java @@ -24,8 +24,10 @@ import io.fabric8.kubernetes.api.model.HasMetadata; import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.DefaultContext; import io.javaoperatorsdk.operator.api.reconciler.Ignore; import io.javaoperatorsdk.operator.api.reconciler.dependent.Deleter; +import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource; import io.javaoperatorsdk.operator.api.reconciler.dependent.ReconcileResult; class BulkDependentResourceReconciler @@ -115,6 +117,14 @@ public R update(R actual, R desired, P primary, Context

context) { return ((Updater) bulkDependentResource).update(actual, desired, primary, context); } + @Override + @SuppressWarnings("unchecked") + protected R getOrComputeDesired(Context

context) { + return ((DefaultContext

) context) + .getOrComputeDesiredStateFor( + this, (DependentResource) bulkDependentResource, p -> desired); + } + @Override public Result match(R resource, P primary, Context

context) { // retrieve the desired state via the context so that it is processed the same way as for diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/PollingDependentResource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/PollingDependentResource.java index 894e359d57..a524fc2ec1 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/PollingDependentResource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/dependent/external/PollingDependentResource.java @@ -55,6 +55,7 @@ protected ExternalResourceCachingEventSource createEventSource( EventSourceContext

context) { return new PollingEventSource<>( resourceType(), + context, new PollingConfiguration<>(name(), this, getPollingPeriod(), resourceIDMapper)); } } 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 d553d14cf9..bd5222fbbb 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 @@ -55,7 +55,9 @@ public class EventSourceManager

public EventSourceManager(Controller

controller) { this( controller, - new EventSources<>(controller.getConfiguration().triggerReconcilerOnAllEvents())); + new EventSources<>( + controller.getConfiguration().triggerReconcilerOnAllEvents(), + () -> controller.getExecutorServiceManager().retryAndRescheduleExecutorService())); } EventSourceManager(Controller

controller, EventSources

eventSources) { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java index b482d26dca..2e8a523de0 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/EventSources.java @@ -23,6 +23,8 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentNavigableMap; import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.ScheduledExecutorService; +import java.util.function.Supplier; import java.util.stream.Stream; import io.fabric8.kubernetes.api.model.HasMetadata; @@ -41,8 +43,17 @@ class EventSources

{ private ControllerEventSource

controllerEventSource; public EventSources(boolean triggerReconcilerOnAllEvents) { + this(triggerReconcilerOnAllEvents, null); + } + + public EventSources( + boolean triggerReconcilerOnAllEvents, + Supplier scheduledExecutorServiceSupplier) { retryAndRescheduleTimerEventSource = - new TimerEventSource<>("RetryAndRescheduleTimerEventSource", triggerReconcilerOnAllEvents); + new TimerEventSource<>( + "RetryAndRescheduleTimerEventSource", + triggerReconcilerOnAllEvents, + scheduledExecutorServiceSupplier); } EventSources() { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingConfiguration.java index 599647ff29..49bc8dd7ba 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingConfiguration.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingConfiguration.java @@ -18,12 +18,15 @@ import java.time.Duration; import java.util.Objects; import java.util.concurrent.ScheduledExecutorService; -import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.function.Predicate; import io.fabric8.kubernetes.api.model.HasMetadata; import io.javaoperatorsdk.operator.processing.ResourceIDMapper; +/** + * @param executorService the executor to run the polls on, {@code null} (the default) to run them + * on the executor the operator shares between all its scheduled tasks + */ public record PerResourcePollingConfiguration( String name, ScheduledExecutorService executorService, @@ -32,6 +35,11 @@ public record PerResourcePollingConfiguration( Predicate

registerPredicate, Duration defaultPollingPeriod) { + /** + * @deprecated not used anymore, polls now run on the executor the operator shares between all its + * scheduled tasks unless an executor is explicitly configured + */ + @Deprecated(forRemoval = true) public static final int DEFAULT_EXECUTOR_THREAD_NUMBER = 1; public PerResourcePollingConfiguration( @@ -42,10 +50,7 @@ public PerResourcePollingConfiguration( Predicate

registerPredicate, Duration defaultPollingPeriod) { this.name = name; - this.executorService = - executorService == null - ? new ScheduledThreadPoolExecutor(DEFAULT_EXECUTOR_THREAD_NUMBER) - : executorService; + this.executorService = executorService; this.resourceIDMapper = resourceIDMapper == null ? ResourceIDMapper.resourceIdProviderMapper() : resourceIDMapper; this.resourceFetcher = Objects.requireNonNull(resourceFetcher); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java index 1ab750d8f0..7faf8832fd 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSource.java @@ -30,7 +30,9 @@ import org.slf4j.LoggerFactory; import io.fabric8.kubernetes.api.model.HasMetadata; +import io.javaoperatorsdk.operator.ExecutorUtilsInternal; import io.javaoperatorsdk.operator.OperatorException; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.Cache; @@ -56,11 +58,17 @@ public class PerResourcePollingEventSource private final Cache

primaryResourceCache; private final Set fetchedForPrimaries = ConcurrentHashMap.newKeySet(); - private final ScheduledExecutorService executorService; + private final ScheduledExecutorService configuredExecutorService; + private final ConfigurationService configurationService; + private final boolean ownsExecutorService; private final ResourceFetcher resourceFetcher; private final Predicate

registerPredicate; private final Duration period; + private final Map registrationsBeforeStart = new ConcurrentHashMap<>(); + + private volatile ScheduledExecutorService executorService; + public PerResourcePollingEventSource( Class resourceClass, EventSourceContext

context, @@ -69,10 +77,60 @@ public PerResourcePollingEventSource( this.primaryResourceCache = context.getPrimaryCache(); this.resourceFetcher = config.resourceFetcher(); this.registerPredicate = config.registerPredicate(); - this.executorService = config.executorService(); + this.configuredExecutorService = config.executorService(); + this.configurationService = configurationServiceOf(context); + // when neither the configuration nor the operator provides one, the event source has to create + // an executor of its own, and is then the one responsible for shutting it down + this.ownsExecutorService = configuredExecutorService == null && configurationService == null; this.period = config.defaultPollingPeriod(); } + /** + * The configuration of the operator the event source belongs to, or {@code null} if it doesn't + * belong to one, which only happens when the event source is used standalone, outside an + * operator. + */ + private static ConfigurationService configurationServiceOf(EventSourceContext context) { + final var controllerConfiguration = context.getControllerConfiguration(); + return controllerConfiguration == null + ? null + : controllerConfiguration.getConfigurationService(); + } + + @Override + public void start() throws OperatorException { + if (isRunning()) { + return; + } + executorService = resolveExecutorService(); + super.start(); + replayRegistrationsBeforeStart(); + } + + private void replayRegistrationsBeforeStart() { + for (var primaryID : registrationsBeforeStart.keySet()) { + var resource = registrationsBeforeStart.remove(primaryID); + if (resource != null) { + checkAndRegisterTask(resource); + } + } + } + + /** + * Resolves the executor to poll on. Note that this happens on every start, and not once at + * creation time, since the operator shuts its executors down when it is stopped and creates new + * ones if it is started again. + */ + private ScheduledExecutorService resolveExecutorService() { + if (configuredExecutorService != null) { + return configuredExecutorService; + } + if (configurationService != null) { + return configurationService.getExecutorServiceManager().scheduledExecutorService(); + } + return ExecutorUtilsInternal.daemonScheduledThreadPool(1, "josdk-polling-" + name()); + } + private Set getAndCacheResource(P primary, boolean fromGetter) { var values = resourceFetcher.fetchResources(primary); var primaryID = ResourceID.fromResource(primary); @@ -92,6 +150,12 @@ private void scheduleNextExecution(P primary, Set actualResources) { executorService.schedule( new FetchingExecutor(primaryID), fetchDuration.toMillis(), TimeUnit.MILLISECONDS); scheduledFutures.put(primaryID, scheduledFuture); + // a fetch that was already running when the event source got stopped can get here after stop + // cleared the map; since the shared executor stays alive, the task has to be withdrawn, + // otherwise the stale entry would prevent the resource from being registered on a restart + if (!isRunning() && scheduledFutures.remove(primaryID, scheduledFuture)) { + scheduledFuture.cancel(true); + } } @Override @@ -107,6 +171,7 @@ public void onResourceUpdated(P newResource, P oldResource) { @Override public void onResourceDeleted(P resource) { var resourceID = ResourceID.fromResource(resource); + registrationsBeforeStart.remove(resourceID); var scheduledFuture = scheduledFutures.remove(resourceID); if (scheduledFuture != null) { log.debug("Canceling scheduledFuture for resource: {}", resource); @@ -121,6 +186,20 @@ public void onResourceDeleted(P resource) { // important because otherwise there will be a race condition related to the timerTasks. private void checkAndRegisterTask(P resource) { var primaryID = ResourceID.fromResource(resource); + // the primary informer is started before this event source, and the executor to poll on is + // only resolved when this one starts, so resources seen in the meantime are registered then + if (!isRunning()) { + registrationsBeforeStart.put(primaryID, resource); + if (!isRunning()) { + return; + } + replayRegistrationsBeforeStart(); + return; + } + registerTask(primaryID, resource); + } + + private synchronized void registerTask(ResourceID primaryID, P resource) { if (scheduledFutures.get(primaryID) == null && (registerPredicate == null || registerPredicate.test(resource))) { var cachedResources = cachedResourcesFor(primaryID); @@ -150,8 +229,13 @@ public void run() { log.warn("No resource in cache for resource ID: {}", primaryID); // no new execution is scheduled in this case, an on delete event should be received shortly } else { - var actualResources = primary.map(p -> getAndCacheResource(p, false)); - scheduleNextExecution(primary.get(), actualResources.orElse(null)); + Set actualResources = null; + try { + actualResources = getAndCacheResource(primary.get(), false); + } catch (RuntimeException e) { + log.error("Error during polling resources for: {}", primaryID, e); + } + scheduleNextExecution(primary.get(), actualResources); } } } @@ -202,6 +286,13 @@ default Optional fetchDelay(Set lastFetchedResource, P primary) { @Override public void stop() throws OperatorException { super.stop(); - executorService.shutdownNow(); + // the tasks have to be cancelled explicitly now that the executor can be shared with the rest + // of the operator, and the map cleared so that they are registered again on a restart + scheduledFutures.values().forEach(future -> future.cancel(true)); + scheduledFutures.clear(); + registrationsBeforeStart.clear(); + if (ownsExecutorService && executorService != null) { + executorService.shutdownNow(); + } } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingConfiguration.java index 9ac1b8cc96..0836058cf4 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingConfiguration.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingConfiguration.java @@ -17,24 +17,40 @@ import java.time.Duration; import java.util.Objects; +import java.util.concurrent.ScheduledExecutorService; import io.javaoperatorsdk.operator.processing.ResourceIDMapper; +/** + * @param executorService the executor to run the polls on, {@code null} (the default) to run them + * on the executor the operator shares between all its scheduled tasks + */ public record PollingConfiguration( String name, PollingEventSource.GenericResourceFetcher genericResourceFetcher, Duration period, - ResourceIDMapper resourceIDMapper) { + ResourceIDMapper resourceIDMapper, + ScheduledExecutorService executorService) { public PollingConfiguration( String name, PollingEventSource.GenericResourceFetcher genericResourceFetcher, Duration period, ResourceIDMapper resourceIDMapper) { + this(name, genericResourceFetcher, period, resourceIDMapper, null); + } + + public PollingConfiguration( + String name, + PollingEventSource.GenericResourceFetcher genericResourceFetcher, + Duration period, + ResourceIDMapper resourceIDMapper, + ScheduledExecutorService executorService) { this.name = name; this.genericResourceFetcher = Objects.requireNonNull(genericResourceFetcher); this.period = period; this.resourceIDMapper = resourceIDMapper == null ? ResourceIDMapper.resourceIdProviderMapper() : resourceIDMapper; + this.executorService = executorService; } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingConfigurationBuilder.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingConfigurationBuilder.java index 0e68876b60..5a12fc4d50 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingConfigurationBuilder.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingConfigurationBuilder.java @@ -16,6 +16,7 @@ package io.javaoperatorsdk.operator.processing.event.source.polling; import java.time.Duration; +import java.util.concurrent.ScheduledExecutorService; import io.javaoperatorsdk.operator.processing.ResourceIDMapper; @@ -24,6 +25,7 @@ public final class PollingConfigurationBuilder { private final PollingEventSource.GenericResourceFetcher genericResourceFetcher; private ResourceIDMapper resourceIDMapper; private String name; + private ScheduledExecutorService executorService; public PollingConfigurationBuilder( PollingEventSource.GenericResourceFetcher fetcher, Duration period) { @@ -42,7 +44,23 @@ public PollingConfigurationBuilder withName(String name) { return this; } + /** + * Runs the polls on the specified executor instead of the one the operator shares between all its + * scheduled tasks. Note that an explicitly provided executor is not managed by the operator: it + * is the caller's responsibility to shut it down. + * + * @param executorService the executor to run the polls on + * @return this builder for chained customization + * @since 5.6.0 + */ + public PollingConfigurationBuilder withExecutorService( + ScheduledExecutorService executorService) { + this.executorService = executorService; + return this; + } + public PollingConfiguration build() { - return new PollingConfiguration<>(name, genericResourceFetcher, period, resourceIDMapper); + return new PollingConfiguration<>( + name, genericResourceFetcher, period, resourceIDMapper, executorService); } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java index 5c85c24f90..de2d7a9bc1 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSource.java @@ -18,15 +18,19 @@ import java.time.Duration; import java.util.Map; import java.util.Set; -import java.util.Timer; -import java.util.TimerTask; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import io.fabric8.kubernetes.api.model.HasMetadata; +import io.javaoperatorsdk.operator.ExecutorUtilsInternal; import io.javaoperatorsdk.operator.OperatorException; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; +import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; import io.javaoperatorsdk.operator.health.Status; import io.javaoperatorsdk.operator.processing.event.ResourceID; import io.javaoperatorsdk.operator.processing.event.source.ExternalResourceCachingEventSource; @@ -62,45 +66,115 @@ public class PollingEventSource private static final Logger log = LoggerFactory.getLogger(PollingEventSource.class); - private Timer timer; private final GenericResourceFetcher genericResourceFetcher; private final Duration period; private final AtomicBoolean healthy = new AtomicBoolean(true); + private final ScheduledExecutorService configuredExecutorService; + private final ConfigurationService configurationService; + private final boolean ownsExecutorService; + private volatile ScheduledExecutorService executorService; + private volatile ScheduledFuture pollingTask; + + /** + * Creates an event source polling on the executor the operator shares between all its scheduled + * tasks, unless the configuration provides one of its own. + * + * @param resourceClass the type of the polled resource + * @param context the context this event source is created for + * @param config the configuration of the polling + * @since 5.6.0 + */ + public PollingEventSource( + Class resourceClass, EventSourceContext

context, PollingConfiguration config) { + this(resourceClass, config, configurationServiceOf(context)); + } + + /** + * @deprecated use {@link #PollingEventSource(Class, EventSourceContext, PollingConfiguration)} + * instead: without a context, and unless the configuration provides an executor, this event + * source has to create a thread of its own to poll on instead of using the one the operator + * shares between all its scheduled tasks. + */ + @Deprecated(since = "5.6.0") public PollingEventSource(Class resourceClass, PollingConfiguration config) { + this(resourceClass, config, null); + } + + private PollingEventSource( + Class resourceClass, + PollingConfiguration config, + ConfigurationService configurationService) { super(config.name(), resourceClass, config.resourceIDMapper()); this.genericResourceFetcher = config.genericResourceFetcher(); this.period = config.period(); + this.configuredExecutorService = config.executorService(); + this.configurationService = configurationService; + // when neither the configuration nor the operator provides one, the event source has to create + // an executor of its own, and is then the one responsible for shutting it down + this.ownsExecutorService = configuredExecutorService == null && configurationService == null; + } + + /** + * The configuration of the operator the event source belongs to, or {@code null} if it doesn't + * belong to one, which only happens when the event source is used standalone, outside an + * operator. + */ + private static ConfigurationService configurationServiceOf(EventSourceContext context) { + final var controllerConfiguration = context.getControllerConfiguration(); + return controllerConfiguration == null + ? null + : controllerConfiguration.getConfigurationService(); } @Override public void start() throws OperatorException { - if (timer != null) { + if (pollingTask != null) { return; } super.start(); + executorService = resolveExecutorService(); getStateAndFillCache(); - timer = new Timer(true); - timer.schedule( - new TimerTask() { - @Override - public void run() { - try { - if (!isRunning()) { - log.debug("Event source not yet started. Will not run."); - return; - } - getStateAndFillCache(); - healthy.set(true); - } catch (Exception e) { - // Exception is required because of Kotlin - healthy.set(false); - log.error("Error during polling.", e); - } - } - }, - period.toMillis(), - period.toMillis()); + var task = + executorService.scheduleWithFixedDelay( + this::poll, period.toMillis(), period.toMillis(), TimeUnit.MILLISECONDS); + pollingTask = task; + // stop() clears the running flag before it looks for the task, so a stop that happened during + // the initial fetch is either seen by stop (which cancels the task) or caught here + if (!isRunning()) { + task.cancel(false); + pollingTask = null; + } + } + + private void poll() { + try { + if (!isRunning()) { + log.debug("Event source not yet started. Will not run."); + return; + } + getStateAndFillCache(); + healthy.set(true); + } catch (Exception e) { + // Exception is required because of Kotlin + healthy.set(false); + log.error("Error during polling.", e); + } + } + + /** + * Resolves the executor to poll on. Note that this happens on every start, and not once at + * creation time, since the operator shuts its executors down when it is stopped and creates new + * ones if it is started again. + */ + private ScheduledExecutorService resolveExecutorService() { + if (configuredExecutorService != null) { + return configuredExecutorService; + } + if (configurationService != null) { + return configurationService.getExecutorServiceManager().scheduledExecutorService(); + } + return ExecutorUtilsInternal.daemonScheduledThreadPool(1, "josdk-polling-" + name()); } protected synchronized void getStateAndFillCache() { @@ -115,9 +189,13 @@ public interface GenericResourceFetcher { @Override public void stop() throws OperatorException { super.stop(); - if (timer != null) { - timer.cancel(); - timer = null; + if (pollingTask != null) { + // as with the java.util.Timer this replaces, an ongoing poll is left to finish + pollingTask.cancel(false); + pollingTask = null; + } + if (ownsExecutorService && executorService != null) { + executorService.shutdownNow(); } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java index eae9663fe6..07de4f0d7e 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSource.java @@ -17,14 +17,20 @@ import java.util.Map; import java.util.Set; -import java.util.Timer; import java.util.TimerTask; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ScheduledFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import io.fabric8.kubernetes.api.model.HasMetadata; +import io.javaoperatorsdk.operator.ExecutorUtilsInternal; import io.javaoperatorsdk.operator.api.reconciler.BaseControl; import io.javaoperatorsdk.operator.health.Status; import io.javaoperatorsdk.operator.processing.event.Event; @@ -36,17 +42,47 @@ public class TimerEventSource extends AbstractEventSource implements ResourceEventAware { private static final Logger log = LoggerFactory.getLogger(TimerEventSource.class); - private Timer timer; - private final Map onceTasks = new ConcurrentHashMap<>(); + private final Map> onceTasks = new ConcurrentHashMap<>(); + private final Supplier executorServiceSupplier; + private final boolean ownsExecutorService; private boolean triggerReconcilerOnAllEvents; + private volatile ScheduledExecutorService executorService; public TimerEventSource() { - super(Void.class); + this((Supplier) null); } public TimerEventSource(String name, boolean triggerReconcilerOnAllEvents) { + this(name, triggerReconcilerOnAllEvents, null); + } + + /** + * Creates an event source scheduling on the executor provided by the specified supplier. The + * supplier is called on every start, and not once at creation time, since the operator shuts its + * executors down when it is stopped and creates new ones if it is started again. + * + * @param executorServiceSupplier supplies the executor to schedule on, {@code null} to have the + * event source create, and shut down, an executor of its own + * @since 5.6.0 + */ + public TimerEventSource(Supplier executorServiceSupplier) { + super(Void.class); + this.executorServiceSupplier = executorServiceSupplier; + this.ownsExecutorService = executorServiceSupplier == null; + } + + /** + * @see #TimerEventSource(Supplier) + * @since 5.6.0 + */ + public TimerEventSource( + String name, + boolean triggerReconcilerOnAllEvents, + Supplier executorServiceSupplier) { super(Void.class, name); this.triggerReconcilerOnAllEvents = triggerReconcilerOnAllEvents; + this.executorServiceSupplier = executorServiceSupplier; + this.ownsExecutorService = executorServiceSupplier == null; } @SuppressWarnings("unused") @@ -55,19 +91,40 @@ public void scheduleOnce(R resource, long delay) { } public void scheduleOnce(ResourceID resourceID, long delay) { - if (!isRunning()) { + final var executor = executorService; + if (!isRunning() || executor == null) { throw new IllegalStateException("The TimerEventSource is not running"); } - if (onceTasks.containsKey(resourceID)) { + if (delay == BaseControl.INSTANT_RESCHEDULE) { cancelOnceSchedule(resourceID); + new EventProducerTimeTask(resourceID).run(); + return; } - EventProducerTimeTask task = new EventProducerTimeTask(resourceID); - if (delay == BaseControl.INSTANT_RESCHEDULE) { - task.run(); - } else { - onceTasks.put(resourceID, task); - timer.schedule(task, delay); + + var scheduled = new AtomicReference>(); + try { + onceTasks.compute( + resourceID, + (id, alreadyScheduled) -> { + if (alreadyScheduled != null) { + alreadyScheduled.cancel(false); + } + var future = + executor.schedule(new EventProducerTimeTask(id), delay, TimeUnit.MILLISECONDS); + scheduled.set(future); + return future; + }); + } catch (RejectedExecutionException e) { + if (isRunning()) { + throw e; + } + throw new IllegalStateException("The TimerEventSource is not running", e); + } + // stop() clears the running flag before cancelling the tasks, so a task that got registered + // concurrently with it is either cancelled by stop, or withdrawn here + if (!isRunning() && onceTasks.remove(resourceID, scheduled.get())) { + scheduled.get().cancel(false); } } @@ -81,26 +138,37 @@ public void onResourceDeleted(R resource) { } public void cancelOnceSchedule(ResourceID customResourceUid) { - TimerTask timerTask = onceTasks.remove(customResourceUid); - if (timerTask != null) { - timerTask.cancel(); + var scheduled = onceTasks.remove(customResourceUid); + if (scheduled != null) { + // as with the java.util.TimerTask this replaces, a task that is already running is left to + // finish + scheduled.cancel(false); } } @Override public void start() { if (!isRunning()) { + executorService = resolveExecutorService(); super.start(); - timer = new Timer(true); } } + private ScheduledExecutorService resolveExecutorService() { + if (executorServiceSupplier != null) { + return executorServiceSupplier.get(); + } + return ExecutorUtilsInternal.daemonScheduledThreadPool(1, "josdk-timer-" + name()); + } + @Override public void stop() { if (isRunning()) { - onceTasks.keySet().forEach(this::cancelOnceSchedule); - timer.cancel(); super.stop(); + onceTasks.keySet().forEach(this::cancelOnceSchedule); + if (ownsExecutorService && executorService != null) { + executorService.shutdownNow(); + } } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/VirtualThreadsStartupDeadlockTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/VirtualThreadsStartupDeadlockTest.java new file mode 100644 index 0000000000..914559eeb8 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/VirtualThreadsStartupDeadlockTest.java @@ -0,0 +1,103 @@ +/* + * 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; + +import java.time.Duration; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.javaoperatorsdk.operator.api.config.BaseConfigurationService; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; +import io.javaoperatorsdk.operator.api.config.MockControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.processing.Controller; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** + * With virtual threads enabled on Java 21 to 24 (no JEP 491, hence only run on Java 25 or later, + * the official baseline for virtual thread support), starting as many controllers as there are + * virtual thread carriers deadlocks, because each controller is started on a virtual thread of the + * caching executor that blocks, while holding the monitors of the synchronized Controller.start / + * EventSourceManager.start / ControllerEventSource.start, on informer-start tasks submitted to the + * very same virtual thread executor. + */ +class VirtualThreadsStartupDeadlockTest { + + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + @SuppressWarnings({"rawtypes", "unchecked"}) + void startingControllersOnVirtualThreadsDoesNotDeadlock() throws Exception { + assertStartsWithoutDeadlock(true); + } + + @Test + void controlStartingControllersOnPlatformThreads() throws Exception { + assertStartsWithoutDeadlock(false); + } + + @SuppressWarnings({"rawtypes", "unchecked"}) + private static void assertStartsWithoutDeadlock(boolean useVirtualThreads) throws Exception { + final var mockClient = MockKubernetesClient.client(ConfigMap.class); + final ConfigurationService configService = + ConfigurationService.newOverriddenConfigurationService( + new BaseConfigurationService(), + o -> o.withKubernetesClient(mockClient).withUseVirtualThreads(useVirtualThreads)); + final var executorServiceManager = configService.getExecutorServiceManager(); + final var controllerManager = new ControllerManager(executorServiceManager); + + // default virtual thread scheduler parallelism == number of available processors, e.g. 1 or 2 + // in a typical resource-limited operator pod + final int controllers = + System.getenv("AUDIT_N") != null + ? Integer.parseInt(System.getenv("AUDIT_N")) + : Runtime.getRuntime().availableProcessors(); + for (int i = 0; i < controllers; i++) { + final var configuration = MockControllerConfiguration.forResource(ConfigMap.class); + when(configuration.getConfigurationService()).thenReturn(configService); + when(configuration.getName()).thenReturn("controller-" + i); + controllerManager.add(new Controller(mock(Reconciler.class), configuration, mockClient)); + } + + final var start = CompletableFuture.runAsync(() -> controllerManager.start(false)); + try { + start.get(20, TimeUnit.SECONDS); + } catch (TimeoutException e) { + // break the deadlock (interrupts the pinned virtual threads) so that the shared virtual + // thread scheduler isn't left starved for the rest of the test JVM + executorServiceManager.stop(Duration.ZERO); + try { + start.get(20, TimeUnit.SECONDS); + } catch (Exception ignored) { + // expected once the stuck start was interrupted + } + assertThat(false) + .as( + "ControllerManager.start deadlocked starting %d controllers on virtual threads", + controllers) + .isTrue(); + } + executorServiceManager.stop(Duration.ZERO); + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/BoundedVirtualThreadExecutorAuditTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/BoundedVirtualThreadExecutorAuditTest.java new file mode 100644 index 0000000000..5e17c52ed9 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/BoundedVirtualThreadExecutorAuditTest.java @@ -0,0 +1,124 @@ +/* + * 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.api.config; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException; + +/** Throwaway audit tests for VirtualThreads.BoundedExecutorService. */ +class BoundedVirtualThreadExecutorAuditTest { + + @Test + void controlPlatformPoolRejectsNonPositiveConcurrency() { + assertThatIllegalArgumentException() + .isThrownBy(() -> ExecutorServiceManager.newBoundedExecutorService(0, false)); + } + + /** + * The platform variant fails fast for a non positive limit, the virtual variant creates a + * Semaphore(0) (or a negative one) whose tasks wait for a permit forever. + */ + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void virtualThreadVariantRejectsNonPositiveConcurrencyToo() { + assertThatIllegalArgumentException() + .isThrownBy(() -> ExecutorServiceManager.newBoundedExecutorService(0, true).shutdownNow()); + } + + @Test + void controlPlatformPoolShutdownNowReturnsTasksThatNeverStarted() throws Exception { + assertThat(queuedTasksReturnedByShutdownNow(false)).isEqualTo(5); + } + + /** + * ExecutorService#shutdownNow must return the tasks that never commenced execution; the bounded + * virtual executor returns an empty list, and plain Runnables handed to execute() that were still + * waiting for a permit are silently dropped (neither run nor returned). + */ + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void virtualThreadVariantShutdownNowReturnsTasksThatNeverStarted() throws Exception { + assertThat(queuedTasksReturnedByShutdownNow(true)).isEqualTo(5); + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void everyTaskSubmittedWhileShuttingDownIsRunReturnedOrRejected() throws Exception { + for (int round = 0; round < 50; round++) { + final var executor = ExecutorServiceManager.newBoundedExecutorService(1, true); + final var accepted = new AtomicInteger(); + final var ran = new AtomicInteger(); + final var submitters = new CountDownLatch(4); + final var go = new CountDownLatch(1); + for (int t = 0; t < 4; t++) { + new Thread( + () -> { + try { + go.await(); + for (int i = 0; i < 200; i++) { + executor.execute(ran::incrementAndGet); + accepted.incrementAndGet(); + } + } catch (RejectedExecutionException | InterruptedException e) { + // the executor was shut down + } finally { + submitters.countDown(); + } + }) + .start(); + } + go.countDown(); + final var notStarted = executor.shutdownNow(); + assertThat(submitters.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + // tasks admitted after shutdownNow returned can't exist, so none may go missing + assertThat(ran.get() + notStarted.size()).isEqualTo(accepted.get()); + } + } + + private static int queuedTasksReturnedByShutdownNow(boolean useVirtualThreads) throws Exception { + final var executor = ExecutorServiceManager.newBoundedExecutorService(1, useVirtualThreads); + final var running = new CountDownLatch(1); + final var ran = new AtomicInteger(); + executor.execute( + () -> { + running.countDown(); + try { + Thread.sleep(60_000); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + }); + assertThat(running.await(10, TimeUnit.SECONDS)).isTrue(); + for (int i = 0; i < 5; i++) { + executor.execute(ran::incrementAndGet); + } + Thread.sleep(200); // let the virtual threads reach the permit acquisition + final var notStarted = executor.shutdownNow(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + assertThat(ran).hasValue(0); + return notStarted.size(); + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java index aec8381135..7acf3fff84 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ConfigurationServiceOverriderTest.java @@ -20,6 +20,7 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor; import org.junit.jupiter.api.Test; @@ -117,6 +118,64 @@ public R clone(R object) { config.reconciliationTerminationTimeout(), overridden.reconciliationTerminationTimeout()); } + @Test + void scheduledExecutorCanBeOverridden() { + final var scheduledExecutorService = Executors.newScheduledThreadPool(1); + final var retryExecutorService = Executors.newScheduledThreadPool(1); + try { + final var overridden = + new ConfigurationServiceOverrider(config) + .withConcurrentScheduledTaskThreads(7) + .withScheduledExecutorService(scheduledExecutorService) + .withConcurrentRetryAndRescheduleThreads(5) + .withRetryAndRescheduleExecutorService(retryExecutorService) + .build(); + + assertThat(overridden.concurrentScheduledTaskThreads()).isEqualTo(7); + assertThat(overridden.getScheduledExecutorService()).isSameAs(scheduledExecutorService); + assertThat(overridden.concurrentRetryAndRescheduleThreads()).isEqualTo(5); + assertThat(overridden.getRetryAndRescheduleExecutorService()).isSameAs(retryExecutorService); + } finally { + scheduledExecutorService.shutdownNow(); + retryExecutorService.shutdownNow(); + } + } + + @Test + void scheduledExecutorDefaultsToADaemonPoolOfTheConfiguredSize() { + final var overridden = + (ScheduledThreadPoolExecutor) + new ConfigurationServiceOverrider(config) + .withConcurrentScheduledTaskThreads(3) + .build() + .getScheduledExecutorService(); + + try { + assertThat(overridden.getCorePoolSize()).isEqualTo(3); + // scheduled tasks must not delay the termination of the operator, nor keep the JVM alive + assertThat(overridden.getExecuteExistingDelayedTasksAfterShutdownPolicy()).isFalse(); + assertThat(overridden.getThreadFactory().newThread(() -> {}).isDaemon()).isTrue(); + } finally { + overridden.shutdownNow(); + } + } + + @Test + void retryAndRescheduleExecutorIsSeparateFromTheScheduledTaskOne() { + final var scheduled = config.getScheduledExecutorService(); + final var retryAndReschedule = config.getRetryAndRescheduleExecutorService(); + + try { + // a slow poll on the scheduled task executor must not be able to delay a retry + assertThat(retryAndReschedule).isNotSameAs(scheduled); + assertThat(((ScheduledThreadPoolExecutor) retryAndReschedule).getCorePoolSize()) + .isEqualTo(ConfigurationService.DEFAULT_RETRY_AND_RESCHEDULE_THREADS_NUMBER); + } finally { + scheduled.shutdownNow(); + retryAndReschedule.shutdownNow(); + } + } + @Test void eventRecorderIsNotConfiguredByDefaultAndCanBeOverridden() { final var eventRecorder = @@ -152,6 +211,13 @@ void threadCountConfiguredProperly() { .isEqualTo(14); } + @Test + void virtualThreadsAreDisabledByDefaultAndCanBeOverridden() { + assertThat(config.useVirtualThreads()).isFalse(); + assertThat(new ConfigurationServiceOverrider(config).withUseVirtualThreads(true).build()) + .returns(true, ConfigurationService::useVirtualThreads); + } + @SuppressWarnings("rawtypes") @Test void dependentResourceFactoryDefaultsToTheSharedOneAndCanBeOverridden() { diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java index 40ffded241..33900d834a 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/ExecutorServiceManagerTest.java @@ -16,10 +16,14 @@ package io.javaoperatorsdk.operator.api.config; import java.time.Duration; +import java.util.Optional; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; class ExecutorServiceManagerTest { @@ -30,14 +34,19 @@ void stopShutsDownTheScheduledExecutorService() { ConfigurationService configurationService = new BaseConfigurationService(); var manager = configurationService.getExecutorServiceManager(); var scheduled = manager.scheduledExecutorService(); + var retryAndReschedule = manager.retryAndRescheduleExecutorService(); try { assertThat(scheduled.isShutdown()).isFalse(); + assertThat(retryAndReschedule.isShutdown()).isFalse(); + // retries are triggered on an executor of their own so that a slow poll can't delay them + assertThat(retryAndReschedule).isNotSameAs(scheduled); } finally { manager.stop(SHUTDOWN_TIMEOUT); } assertThat(scheduled.isShutdown()).isTrue(); + assertThat(retryAndReschedule.isShutdown()).isTrue(); } @Test @@ -53,7 +62,55 @@ void canBeRestartedAfterStop() { assertThat(manager.reconcileExecutorService().isShutdown()).isFalse(); assertThat(manager.cachingExecutorService().isShutdown()).isFalse(); assertThat(manager.scheduledExecutorService().isShutdown()).isFalse(); + assertThat(manager.retryAndRescheduleExecutorService().isShutdown()).isFalse(); manager.stop(SHUTDOWN_TIMEOUT); } + + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void cachingExecutorRunsOnVirtualThreadsWhenRequested() throws Exception { + var configurationService = + ConfigurationService.newOverriddenConfigurationService( + new BaseConfigurationService(), o -> o.withUseVirtualThreads(true)); + var manager = configurationService.getExecutorServiceManager(); + try { + var virtual = + manager.cachingExecutorService().submit(() -> VirtualThreadsTest.onVirtualThread()).get(); + assertThat(virtual).isTrue(); + } finally { + manager.stop(SHUTDOWN_TIMEOUT); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void keepAliveThreadRunsOnlyWhileTheManagerIsStartedWithVirtualThreads() throws Exception { + var configurationService = + ConfigurationService.newOverriddenConfigurationService( + new BaseConfigurationService(), o -> o.withUseVirtualThreads(true)); + var manager = configurationService.getExecutorServiceManager(); + + assertThat(keepAliveThread()).hasValueSatisfying(t -> assertThat(t.isDaemon()).isFalse()); + + manager.stop(SHUTDOWN_TIMEOUT); + + await().untilAsserted(() -> assertThat(keepAliveThread()).isEmpty()); + } + + @Test + void noKeepAliveThreadWithPlatformThreads() { + var manager = new BaseConfigurationService().getExecutorServiceManager(); + try { + assertThat(keepAliveThread()).isEmpty(); + } finally { + manager.stop(SHUTDOWN_TIMEOUT); + } + } + + private static Optional keepAliveThread() { + return Thread.getAllStackTraces().keySet().stream() + .filter(t -> t.getName().equals("josdk-keep-alive") && t.isAlive()) + .findFirst(); + } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadScheduledExecutorTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadScheduledExecutorTest.java new file mode 100644 index 0000000000..ae9f617863 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadScheduledExecutorTest.java @@ -0,0 +1,276 @@ +/* + * 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.api.config; + +import java.time.Duration; +import java.util.concurrent.CancellationException; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatExceptionOfType; +import static org.awaitility.Awaitility.await; + +@EnabledForJreRange(min = JRE.JAVA_25) +class VirtualThreadScheduledExecutorTest { + + private static final int MAX_CONCURRENCY = 2; + + private ScheduledExecutorService executor; + + @BeforeEach + void setUp() { + executor = ExecutorServiceManager.newScheduledExecutorService(MAX_CONCURRENCY, true, "test"); + } + + @AfterEach + void tearDown() { + executor.shutdownNow(); + } + + @Test + void runsDelayedTasksOnVirtualThreads() throws Exception { + var future = executor.schedule(VirtualThreadsTest::onVirtualThread, 50, TimeUnit.MILLISECONDS); + + assertThat(future.get(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void keepsTheTimingOnADaemonPlatformThread() { + executor.schedule(() -> {}, 10, TimeUnit.MINUTES); + + assertThat(Thread.getAllStackTraces().keySet()) + .filteredOn(t -> t.getName().startsWith("test-timer")) + .isNotEmpty() + .allSatisfy(t -> assertThat(t.isDaemon()).isTrue()); + } + + @Test + void delayedTasksDoNotRunEarly() throws Exception { + var start = System.nanoTime(); + + executor.schedule(() -> {}, 200, TimeUnit.MILLISECONDS).get(10, TimeUnit.SECONDS); + + assertThat(Duration.ofNanos(System.nanoTime() - start)) + .isGreaterThanOrEqualTo(Duration.ofMillis(190)); + } + + @Test + void cancellingBeforeTheTaskFiresPreventsIt() throws Exception { + var ran = new AtomicBoolean(); + var future = executor.schedule(() -> ran.set(true), 200, TimeUnit.MILLISECONDS); + + assertThat(future.cancel(false)).isTrue(); + + assertThat(future.isCancelled()).isTrue(); + assertThatExceptionOfType(CancellationException.class).isThrownBy(future::get); + Thread.sleep(400); + assertThat(ran).isFalse(); + } + + @Test + void propagatesTheFailureOfATask() { + var future = + executor.schedule( + () -> { + throw new IllegalStateException("failed"); + }, + 0, + TimeUnit.MILLISECONDS); + + assertThatExceptionOfType(ExecutionException.class) + .isThrownBy(() -> future.get(10, TimeUnit.SECONDS)) + .withCauseInstanceOf(IllegalStateException.class); + } + + @Test + void repeatsTasksWithAFixedDelayUntilCancelled() { + var runs = new AtomicInteger(); + var onVirtualThread = new AtomicBoolean(true); + + var future = + executor.scheduleWithFixedDelay( + () -> { + onVirtualThread.compareAndSet(true, VirtualThreadsTest.onVirtualThread()); + runs.incrementAndGet(); + }, + 0, + 20, + TimeUnit.MILLISECONDS); + + await() + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> assertThat(runs).hasValueGreaterThan(3)); + assertThat(future.cancel(false)).isTrue(); + var runsAtCancel = runs.get(); + await() + .pollDelay(Duration.ofMillis(200)) + .untilAsserted(() -> assertThat(runs.get()).isLessThanOrEqualTo(runsAtCancel + 1)); + assertThat(onVirtualThread).isTrue(); + } + + @Test + void repeatsTasksAtAFixedRate() { + var runs = new AtomicInteger(); + + executor.scheduleAtFixedRate(runs::incrementAndGet, 0, 20, TimeUnit.MILLISECONDS); + + await() + .atMost(Duration.ofSeconds(10)) + .untilAsserted(() -> assertThat(runs).hasValueGreaterThan(3)); + } + + @Test + void aFailingPeriodicTaskIsNotRepeated() { + var runs = new AtomicInteger(); + + var future = + executor.scheduleWithFixedDelay( + () -> { + runs.incrementAndGet(); + throw new IllegalStateException("failed"); + }, + 0, + 20, + TimeUnit.MILLISECONDS); + + assertThatExceptionOfType(ExecutionException.class) + .isThrownBy(() -> future.get(10, TimeUnit.SECONDS)); + await().pollDelay(Duration.ofMillis(200)).untilAsserted(() -> assertThat(runs).hasValue(1)); + } + + @Test + void runsAtMostTheConfiguredNumberOfTasksAtTheSameTime() throws Exception { + var running = new AtomicInteger(); + var maxRunning = new AtomicInteger(); + var done = new CountDownLatch(10); + for (int i = 0; i < 10; i++) { + executor.schedule( + () -> { + maxRunning.accumulateAndGet(running.incrementAndGet(), Math::max); + try { + Thread.sleep(30); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + running.decrementAndGet(); + done.countDown(); + }, + 0, + TimeUnit.MILLISECONDS); + } + + assertThat(done.await(30, TimeUnit.SECONDS)).isTrue(); + assertThat(maxRunning).hasValueLessThanOrEqualTo(MAX_CONCURRENCY); + } + + @Test + void shutdownDiscardsTasksScheduledForLaterAndRejectsNewOnes() throws Exception { + var ran = new AtomicBoolean(); + var later = executor.schedule(() -> ran.set(true), 10, TimeUnit.MINUTES); + var periodic = executor.scheduleWithFixedDelay(() -> {}, 10, 10, TimeUnit.MINUTES); + + executor.shutdown(); + + assertThat(later.isCancelled()).isTrue(); + assertThat(periodic.isCancelled()).isTrue(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + assertThat(ran).isFalse(); + assertThatExceptionOfType(RejectedExecutionException.class) + .isThrownBy(() -> executor.schedule(() -> {}, 0, TimeUnit.MILLISECONDS)); + } + + @Test + void shutdownLetsStartedTasksFinish() throws Exception { + var started = new CountDownLatch(1); + var finished = new AtomicBoolean(); + var future = + executor.schedule( + () -> { + started.countDown(); + Thread.sleep(200); + finished.set(true); + return true; + }, + 0, + TimeUnit.MILLISECONDS); + assertThat(started.await(10, TimeUnit.SECONDS)).isTrue(); + + executor.shutdown(); + + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + assertThat(future.get()).isTrue(); + assertThat(finished).isTrue(); + } + + @Test + void shutdownNowInterruptsRunningTasks() throws Exception { + var started = new CountDownLatch(1); + var interrupted = new CountDownLatch(1); + executor.schedule( + () -> { + started.countDown(); + try { + Thread.sleep(60_000); + } catch (InterruptedException e) { + interrupted.countDown(); + } + }, + 0, + TimeUnit.MILLISECONDS); + assertThat(started.await(10, TimeUnit.SECONDS)).isTrue(); + + executor.shutdownNow(); + + assertThat(interrupted.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(executor.awaitTermination(10, TimeUnit.SECONDS)).isTrue(); + } + + @Test + void theOperatorsDefaultScheduledExecutorsRunOnVirtualThreadsWhenRequested() throws Exception { + var configurationService = + ConfigurationService.newOverriddenConfigurationService( + new BaseConfigurationService(), o -> o.withUseVirtualThreads(true)); + var manager = configurationService.getExecutorServiceManager(); + try { + assertThat( + manager + .scheduledExecutorService() + .schedule(VirtualThreadsTest::onVirtualThread, 0, TimeUnit.MILLISECONDS) + .get(10, TimeUnit.SECONDS)) + .isTrue(); + assertThat( + manager + .retryAndRescheduleExecutorService() + .schedule(VirtualThreadsTest::onVirtualThread, 0, TimeUnit.MILLISECONDS) + .get(10, TimeUnit.SECONDS)) + .isTrue(); + } finally { + manager.stop(Duration.ofMillis(100)); + } + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsJvmLifetimeTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsJvmLifetimeTest.java new file mode 100644 index 0000000000..6ff0cc32fd --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsJvmLifetimeTest.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.api.config; + +import java.nio.file.Path; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Throwaway audit test: the usual operator main method ends right after {@code operator.start()} + * (see docs/.../operations/configuration.md and several samples) and relies on the framework's + * non-daemon executor threads to keep the JVM alive. Virtual threads are always daemon threads, so + * with virtual threads enabled nothing keeps the JVM running once main returns. + */ +class VirtualThreadsJvmLifetimeTest { + + /** Mimics an operator main: start the executors, run a reconciliation, return from main. */ + public static class OperatorLikeMain { + public static void main(String[] args) throws Exception { + final boolean useVirtualThreads = Boolean.parseBoolean(args[0]); + final var configService = + ConfigurationService.newOverriddenConfigurationService( + new BaseConfigurationService(), o -> o.withUseVirtualThreads(useVirtualThreads)); + final var esm = configService.getExecutorServiceManager(); + // what Operator.start() does: start informers on the caching executor ... + esm.cachingExecutorService().submit(() -> {}).get(); + // ... and a first reconciliation dispatched by the EventProcessor + esm.reconcileExecutorService().submit(() -> {}).get(); + System.out.println("main returning"); + } + } + + private static boolean jvmStillAliveAfterMainReturns(boolean useVirtualThreads) throws Exception { + final var java = Path.of(System.getProperty("java.home"), "bin", "java").toString(); + final var process = + new ProcessBuilder( + java, + "-cp", + System.getProperty("java.class.path"), + OperatorLikeMain.class.getName(), + Boolean.toString(useVirtualThreads)) + .redirectErrorStream(true) + .start(); + try { + final boolean exited = process.waitFor(10, TimeUnit.SECONDS); + if (exited) { + final var output = new String(process.getInputStream().readAllBytes()); + // make sure the child exited normally after main returned, not because it crashed + assertThat(output).contains("main returning"); + assertThat(process.exitValue()).isZero(); + } + return !exited; + } finally { + process.destroyForcibly(); + } + } + + @Test + void controlPlatformThreadsKeepTheOperatorJvmAlive() throws Exception { + assertThat(jvmStillAliveAfterMainReturns(false)).isTrue(); + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void virtualThreadsKeepTheOperatorJvmAliveToo() throws Exception { + assertThat(jvmStillAliveAfterMainReturns(true)) + .as("JVM exited right after main returned with useVirtualThreads=true") + .isTrue(); + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java new file mode 100644 index 0000000000..6c3b0854e3 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java @@ -0,0 +1,221 @@ +/* + * 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.api.config; + +import java.util.Set; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.stream.IntStream; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.condition.EnabledForJreRange; +import org.junit.jupiter.api.condition.JRE; + +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.impl.BaseClient; + +import static org.assertj.core.api.Assertions.assertThat; + +class VirtualThreadsTest { + + private static final int MAX_CONCURRENCY = 3; + private static final int TASK_NUMBER = 20; + private static final int TIMEOUT_SECONDS = 30; + + @Test + void usesPlatformThreadPoolWhenVirtualThreadsAreNotRequested() throws Exception { + var executor = ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, false); + try { + assertThat(executor).isInstanceOf(ThreadPoolExecutor.class); + assertThat(executor.submit(VirtualThreadsTest::onVirtualThread).get()).isFalse(); + } finally { + executor.shutdownNow(); + } + } + + @Test + void virtualThreadsAreOnlySupportedFromJava25() { + assertThat(VirtualThreads.isSupported(17)).isFalse(); + assertThat(VirtualThreads.isSupported(21)).isFalse(); + assertThat(VirtualThreads.isSupported(24)).isFalse(); + assertThat(VirtualThreads.isSupported(Runtime.version().feature())) + .isEqualTo(Runtime.version().feature() >= 25); + } + + @Test + @EnabledForJreRange(max = JRE.JAVA_24) + void fallsBackToPlatformThreadsBeforeJava25() throws Exception { + var executor = ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, true); + var scheduled = ExecutorServiceManager.newScheduledExecutorService(MAX_CONCURRENCY, true, "t"); + try { + assertThat(executor).isInstanceOf(ThreadPoolExecutor.class); + assertThat(executor.submit(VirtualThreadsTest::onVirtualThread).get()).isFalse(); + assertThat(scheduled.submit(VirtualThreadsTest::onVirtualThread).get()).isFalse(); + } finally { + executor.shutdownNow(); + scheduled.shutdownNow(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void usesVirtualThreadsWhenRequested() throws Exception { + var virtualExecutor = ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, true); + try { + assertThat(virtualExecutor.submit(VirtualThreadsTest::onVirtualThread).get()).isTrue(); + } finally { + virtualExecutor.shutdownNow(); + } + + var unbounded = ExecutorServiceManager.newUnboundedExecutorService(true); + try { + assertThat(unbounded.submit(VirtualThreadsTest::onVirtualThread).get()).isTrue(); + } finally { + unbounded.shutdownNow(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void boundedVirtualThreadExecutorRespectsTheConfiguredConcurrency() throws Exception { + var executor = ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, true); + try { + final var running = new AtomicInteger(); + final var maxObservedConcurrency = new AtomicInteger(); + final var done = new CountDownLatch(TASK_NUMBER); + // each task gets its own (virtual) thread, only the number of concurrently running ones is + // capped + final Set usedThreads = ConcurrentHashMap.newKeySet(); + + IntStream.range(0, TASK_NUMBER) + .forEach( + i -> + executor.execute( + () -> { + usedThreads.add(Thread.currentThread()); + maxObservedConcurrency.accumulateAndGet( + running.incrementAndGet(), Math::max); + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + running.decrementAndGet(); + done.countDown(); + } + })); + + assertThat(done.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)).isTrue(); + assertThat(maxObservedConcurrency).hasValue(MAX_CONCURRENCY); + assertThat(usedThreads).hasSize(TASK_NUMBER); + } finally { + executor.shutdownNow(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void invokeAllIsBoundedTooSinceItIsUsedToStartTheEventSources() { + var executor = ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, true); + try { + final var running = new AtomicInteger(); + final var maxObservedConcurrency = new AtomicInteger(); + + ExecutorServiceManager.executeAndWaitForAllToComplete( + IntStream.range(0, TASK_NUMBER).boxed(), + i -> { + maxObservedConcurrency.accumulateAndGet(running.incrementAndGet(), Math::max); + try { + Thread.sleep(50); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + running.decrementAndGet(); + } + return null; + }, + i -> "task-" + i, + executor); + + assertThat(maxObservedConcurrency).hasValue(MAX_CONCURRENCY); + } finally { + executor.shutdownNow(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void shutdownTerminatesOnceTheAlreadySubmittedTasksAreDone() throws Exception { + ExecutorService executor = + ExecutorServiceManager.newBoundedExecutorService(MAX_CONCURRENCY, true); + final var done = new CountDownLatch(TASK_NUMBER); + + IntStream.range(0, TASK_NUMBER).forEach(i -> executor.execute(done::countDown)); + executor.shutdown(); + + assertThat(executor.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS)).isTrue(); + assertThat(executor.isShutdown()).isTrue(); + assertThat(done.getCount()).isZero(); + } + + @Test + void defaultKubernetesClientKeepsPlatformThreadsWhenVirtualThreadsAreNotRequested() + throws Exception { + try (var client = defaultKubernetesClient(false)) { + assertThat(runsTasksOnVirtualThread(client)).isFalse(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_25) + void defaultKubernetesClientRunsItsTasksOnVirtualThreadsWhenRequested() throws Exception { + ExecutorService executor; + try (var client = defaultKubernetesClient(true)) { + assertThat(runsTasksOnVirtualThread(client)).isTrue(); + executor = (ExecutorService) client.adapt(BaseClient.class).getExecutor(); + } + assertThat(executor.isShutdown()).isTrue(); + } + + private static KubernetesClient defaultKubernetesClient(boolean useVirtualThreads) { + return ConfigurationService.newOverriddenConfigurationService( + o -> o.withUseVirtualThreads(useVirtualThreads)) + .getKubernetesClient(); + } + + private static boolean runsTasksOnVirtualThread(KubernetesClient client) throws Exception { + return CompletableFuture.supplyAsync( + VirtualThreadsTest::onVirtualThread, client.adapt(BaseClient.class).getExecutor()) + .get(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } + + /** + * {@code Thread.isVirtual} only exists as of Java 21 while the tests are compiled for Java 17, + * hence the reflective call. + */ + static boolean onVirtualThread() { + try { + return (Boolean) Thread.class.getMethod("isVirtual").invoke(Thread.currentThread()); + } catch (ReflectiveOperationException e) { + return false; + } + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/DesiredStateAspectAuditTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/DesiredStateAspectAuditTest.java new file mode 100644 index 0000000000..595327ec50 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/DesiredStateAspectAuditTest.java @@ -0,0 +1,222 @@ +/* + * 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.dependent; + +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; + +import org.junit.jupiter.api.Test; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.ConfigMapBuilder; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; +import io.javaoperatorsdk.operator.api.config.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.DefaultContext; +import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource; +import io.javaoperatorsdk.operator.api.reconciler.dependent.DesiredStateAspect; +import io.javaoperatorsdk.operator.processing.Controller; +import io.javaoperatorsdk.operator.processing.dependent.Matcher.Result; +import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource; +import io.javaoperatorsdk.operator.processing.event.EventSourceManager; +import io.javaoperatorsdk.operator.processing.event.source.EventSource; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatCode; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +/** Audit repro tests for the DesiredStateAspect feature (commit 110199c47). */ +@SuppressWarnings({"unchecked", "rawtypes"}) +class DesiredStateAspectAuditTest { + + private static final ConfigMap PRIMARY = + new ConfigMapBuilder() + .withNewMetadata() + .withName("primary") + .withNamespace("ns") + .endMetadata() + .build(); + + private static Controller controllerWith(List aspects) { + final ConfigurationService configurationService = mock(); + when(configurationService.desiredStateAspects()).thenReturn(aspects); + final ControllerConfiguration controllerConfiguration = mock(); + when(controllerConfiguration.getConfigurationService()).thenReturn(configurationService); + final Controller controller = mock(); + when(controller.getConfiguration()).thenReturn(controllerConfiguration); + return controller; + } + + /** + * The aspect contract says it receives "the DependentResource the desired state was computed + * for". For bulk dependents, it instead receives the private per-item wrapper + * BulkDependentResourceReconciler.BulkDependentResourceInstance, so aspects keyed on the + * dependent (type checks, name(), getClass().getSimpleName() as in DesiredStateAspectIT) + * misbehave. + */ + @Test + void bulkDependentAspectReceivesTheUserBulkDependentResource() { + final List> seen = new ArrayList<>(); + final DesiredStateAspect aspect = + (desired, dependentResource, ctx) -> seen.add(dependentResource); + final var context = + new DefaultContext<>(null, controllerWith(List.of(aspect)), PRIMARY, false, false); + + final var bulk = new CreatingBulk(); + bulk.reconcile(PRIMARY, context); + + assertThat(seen).hasSize(1); + assertThat(seen.get(0).name()).isEqualTo(bulk.name()); + assertThat(seen.get(0)).isSameAs(bulk); + } + + /** + * Aspects run inside ConcurrentHashMap.computeIfAbsent. An aspect that looks up the actual + * resource of the dependent it is given (e.g. to preserve an existing label) re-enters + * getOrComputeDesired for the same key through KubernetesDependentResource + * .selectTargetSecondaryResource -> targetSecondaryResourceID, and CHM throws "Recursive update". + */ + @Test + void aspectCanLookUpActualResourceOfTheDependent() { + final DesiredStateAspect aspect = + (desired, dependentResource, ctx) -> + ((DependentResource) dependentResource) + .getSecondaryResource(ctx.getPrimaryResource(), ctx); + final var controller = controllerWith(List.of(aspect)); + final EventSourceManager esm = mock(); + final EventSource es = mock(); + when(es.getSecondaryResources(any())).thenReturn(Set.of(configMap("child", "actual"))); + when(esm.getEventSourcesFor(ConfigMap.class)).thenReturn(List.of(es)); + when(controller.getEventSourceManager()).thenReturn(esm); + final var context = new DefaultContext<>(null, controller, PRIMARY, false, false); + + final var dependent = new ChildConfigMapDependent(); + + assertThatCode(() -> dependent.getSecondaryResource(PRIMARY, context)) + .doesNotThrowAnyException(); + } + + /** + * Before 110199c47, a DefaultContext built on a Controller whose getConfiguration() is not + * stubbed (the pattern the existing tests used: new DefaultContext<>(mock(), mock(), ...)) could + * compute desired states. Now applyDesiredStateAspects dereferences the null + * ControllerConfiguration for every HasMetadata desired state. + */ + @Test + void desiredComputationDoesNotRequireAspectConfiguration() { + final var context = new DefaultContext<>(null, mock(Controller.class), PRIMARY, false, false); + final var dependent = new ChildConfigMapDependent(); + + assertThatCode(() -> dependent.getOrComputeDesired(context)).doesNotThrowAnyException(); + } + + private static ConfigMap configMap(String name, String value) { + return new ConfigMapBuilder() + .withNewMetadata() + .withName(name) + .withNamespace("ns") + .endMetadata() + .withData(Map.of("key", value)) + .build(); + } + + private static class ChildConfigMapDependent + extends KubernetesDependentResource { + ChildConfigMapDependent() { + super(ConfigMap.class); + } + + @Override + protected ConfigMap desired(ConfigMap primary, Context context) { + return configMap("child", "desired"); + } + + @Override + public ConfigMap getOrComputeDesired(Context context) { + return super.getOrComputeDesired(context); + } + } + + private static class CreatingBulk extends AbstractDependentResource + implements BulkDependentResource, + Creator { + + CreatingBulk() { + super("my-bulk"); + } + + @Override + public Map desiredResources(ConfigMap primary, Context context) { + var desired = new LinkedHashMap(); + desired.put("a", configMap("a", "desired")); + return desired; + } + + @Override + public Map getSecondaryResources( + ConfigMap primary, Context context) { + return Map.of(); + } + + @Override + public ConfigMap create(ConfigMap desired, ConfigMap primary, Context context) { + return desired; + } + + @Override + public void deleteTargetResource( + ConfigMap primary, ConfigMap resource, String key, Context context) {} + + @Override + public Result match( + ConfigMap actualResource, + ConfigMap desired, + ConfigMap primary, + Context context) { + return Result.computed(false, desired); + } + + @Override + public Result match( + ConfigMap resource, ConfigMap primary, Context context) { + return Result.computed(false, resource); + } + + @Override + protected Optional selectTargetSecondaryResource( + Set secondaryResources, ConfigMap primary, Context context) { + return Optional.empty(); + } + + @Override + protected void onCreated(ConfigMap primary, ConfigMap created, Context context) {} + + @Override + protected void onUpdated( + ConfigMap primary, ConfigMap updated, ConfigMap actual, Context context) {} + + @Override + public Class resourceType() { + return ConfigMap.class; + } + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSourceTest.java index 907ed19fe5..cb9b82e93c 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PerResourcePollingEventSourceTest.java @@ -24,6 +24,8 @@ import org.junit.jupiter.api.Test; import io.javaoperatorsdk.operator.TestUtils; +import io.javaoperatorsdk.operator.api.config.BaseConfigurationService; +import io.javaoperatorsdk.operator.api.config.ControllerConfiguration; import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; import io.javaoperatorsdk.operator.processing.ResourceIDMapper; import io.javaoperatorsdk.operator.processing.event.EventHandler; @@ -88,6 +90,64 @@ void pollsTheResourceAfterAwareOfIt() { }); } + @Test + void keepsPollingAfterTheFetcherFailed() { + when(supplier.fetchResources(any())) + .thenThrow(new IllegalStateException("temporary failure")) + .thenReturn(Set.of(SampleExternalResource.testResource1())); + + source.onResourceCreated(testCustomResource); + + await() + .untilAsserted(() -> verify(supplier, atLeast(3)).fetchResources(eq(testCustomResource))); + } + + @Test + void resourcesSeenBeforeStartArePolledOnceStarted() { + source.stop(); + setUpSource( + new PerResourcePollingEventSource<>( + SampleExternalResource.class, + context, + new PerResourcePollingConfigurationBuilder< + SampleExternalResource, TestCustomResource, String>( + supplier, Duration.ofMillis(PERIOD)) + .withResourceIDMapper(r -> r.getName() + "#" + r.getValue()) + .build()), + false); + + source.onResourceCreated(testCustomResource); + verify(supplier, never()).fetchResources(any()); + + source.start(); + + await() + .untilAsserted(() -> verify(supplier, atLeast(1)).fetchResources(eq(testCustomResource))); + } + + @Test + void resourcesDeletedBeforeStartAreNotPolled() { + source.stop(); + setUpSource( + new PerResourcePollingEventSource<>( + SampleExternalResource.class, + context, + new PerResourcePollingConfigurationBuilder< + SampleExternalResource, TestCustomResource, String>( + supplier, Duration.ofMillis(PERIOD)) + .withResourceIDMapper(r -> r.getName() + "#" + r.getValue()) + .build()), + false); + + source.onResourceCreated(testCustomResource); + source.onResourceDeleted(testCustomResource); + source.start(); + + await() + .pollDelay(Duration.ofMillis(2 * PERIOD)) + .untilAsserted(() -> verify(supplier, never()).fetchResources(any())); + } + @Test void registeringTaskOnAPredicate() { setUpSource( @@ -215,6 +275,58 @@ void supportsDynamicPollingDelay() { .untilAsserted(() -> verify(supplier, times(2)).fetchResources(any())); } + @Test + void pollsAgainAfterRestart() { + source.onResourceCreated(testCustomResource); + await().untilAsserted(() -> verify(supplier, atLeast(2)).fetchResources(any())); + source.stop(); + + source.start(); + clearInvocations(supplier); + source.onResourceCreated(testCustomResource); + + await().untilAsserted(() -> verify(supplier, atLeast(2)).fetchResources(any())); + } + + @Test + void pollsOnTheNewSharedSchedulerWhenTheOperatorIsRestarted() { + var configurationService = new BaseConfigurationService(); + var executorServiceManager = configurationService.getExecutorServiceManager(); + var controllerConfiguration = mock(ControllerConfiguration.class); + when(controllerConfiguration.getConfigurationService()).thenReturn(configurationService); + when(context.getControllerConfiguration()).thenReturn(controllerConfiguration); + setUpSource( + new PerResourcePollingEventSource<>( + SampleExternalResource.class, + context, + new PerResourcePollingConfigurationBuilder< + SampleExternalResource, TestCustomResource, String>( + supplier, Duration.ofMillis(PERIOD)) + .withResourceIDMapper(r -> r.getName() + "#" + r.getValue()) + .build())); + + try { + source.onResourceCreated(testCustomResource); + await().untilAsserted(() -> verify(supplier, atLeast(2)).fetchResources(any())); + var firstExecutor = executorServiceManager.scheduledExecutorService(); + source.stop(); + + executorServiceManager.stop(Duration.ofMillis(100)); + executorServiceManager.start(configurationService); + assertThat(firstExecutor.isShutdown()).isTrue(); + assertThat(executorServiceManager.scheduledExecutorService()).isNotSameAs(firstExecutor); + + source.start(); + clearInvocations(supplier); + source.onResourceCreated(testCustomResource); + + await().untilAsserted(() -> verify(supplier, atLeast(2)).fetchResources(any())); + } finally { + source.stop(); + executorServiceManager.stop(Duration.ofMillis(100)); + } + } + @Test void deleteEventCancelsTheScheduling() { when(supplier.fetchResources(any())).thenReturn(Set.of(SampleExternalResource.testResource1())); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java index 92400df4de..c9be8caf58 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/polling/PollingEventSourceTest.java @@ -19,11 +19,16 @@ import java.util.HashMap; import java.util.Map; import java.util.Set; +import java.util.concurrent.CopyOnWriteArrayList; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import io.fabric8.kubernetes.api.model.HasMetadata; +import io.javaoperatorsdk.operator.api.config.BaseConfigurationService; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; +import io.javaoperatorsdk.operator.api.config.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; import io.javaoperatorsdk.operator.health.Status; import io.javaoperatorsdk.operator.processing.event.EventHandler; import io.javaoperatorsdk.operator.processing.event.ResourceID; @@ -46,6 +51,7 @@ class PollingEventSourceTest private final PollingEventSource.GenericResourceFetcher resourceFetcher = mock(PollingEventSource.GenericResourceFetcher.class); + @SuppressWarnings("deprecation") private final PollingEventSource pollingEventSource = new PollingEventSource<>( SampleExternalResource.class, @@ -67,15 +73,61 @@ void canBeRestartedAfterStop() throws InterruptedException { Thread.sleep(DEFAULT_WAIT_PERIOD); pollingEventSource.stop(); - // a cancelled java.util.Timer cannot be reused, a new one has to be created on start + // the polling task is cancelled on stop, a new one has to be scheduled on start pollingEventSource.start(); - Thread.sleep(DEFAULT_WAIT_PERIOD); + clearInvocations(resourceFetcher); + await().untilAsserted(() -> verify(resourceFetcher, atLeast(2)).fetchResources()); assertThat(pollingEventSource.getStatus()).isEqualTo(Status.HEALTHY); } @Test - void timerThreadIsADaemonSoItDoesNotKeepTheJvmAlive() throws InterruptedException { + void pollsOnTheNewSharedSchedulerWhenTheOperatorIsRestarted() { + when(resourceFetcher.fetchResources()).thenReturn(testResponseWithOneValue()); + var configurationService = new BaseConfigurationService(); + var executorServiceManager = configurationService.getExecutorServiceManager(); + var eventSource = + new PollingEventSource( + SampleExternalResource.class, + contextFor(configurationService), + new PollingConfiguration<>(null, resourceFetcher, POLL_PERIOD, null)); + eventSource.setEventHandler(mock(EventHandler.class)); + + try { + eventSource.start(); + var firstExecutor = executorServiceManager.scheduledExecutorService(); + eventSource.stop(); + + executorServiceManager.stop(Duration.ofMillis(100)); + executorServiceManager.start(configurationService); + assertThat(firstExecutor.isShutdown()).isTrue(); + assertThat(executorServiceManager.scheduledExecutorService()).isNotSameAs(firstExecutor); + + eventSource.start(); + clearInvocations(resourceFetcher); + + await().untilAsserted(() -> verify(resourceFetcher, atLeast(2)).fetchResources()); + } finally { + eventSource.stop(); + executorServiceManager.stop(Duration.ofMillis(100)); + } + } + + @Test + void stopCancelsThePollingTask() throws InterruptedException { + when(resourceFetcher.fetchResources()).thenReturn(testResponseWithTwoValues()); + pollingEventSource.start(); + Thread.sleep(DEFAULT_WAIT_PERIOD); + pollingEventSource.stop(); + clearInvocations(resourceFetcher); + + Thread.sleep(DEFAULT_WAIT_PERIOD); + + verify(resourceFetcher, never()).fetchResources(); + } + + @Test + void pollingThreadIsADaemonSoItDoesNotKeepTheJvmAlive() throws InterruptedException { when(resourceFetcher.fetchResources()).thenReturn(testResponseWithTwoValues()); var threadsBeforeStart = Thread.getAllStackTraces().keySet(); @@ -83,13 +135,13 @@ void timerThreadIsADaemonSoItDoesNotKeepTheJvmAlive() throws InterruptedExceptio pollingEventSource.start(); Thread.sleep(DEFAULT_WAIT_PERIOD); - var newTimerThreads = + var newPollingThreads = Thread.getAllStackTraces().keySet().stream() - .filter(t -> t.getName().startsWith("Timer-")) + .filter(t -> t.getName().startsWith("josdk-polling-")) .filter(t -> !threadsBeforeStart.contains(t)) .toList(); - assertThat(newTimerThreads).isNotEmpty().allMatch(Thread::isDaemon); + assertThat(newPollingThreads).isNotEmpty().allMatch(Thread::isDaemon); } @Test @@ -154,6 +206,54 @@ void updatesHealthIndicatorBasedOnExceptionsInFetcher() { .untilAsserted(() -> assertThat(pollingEventSource.getStatus()).isEqualTo(Status.HEALTHY)); } + @Test + void pollsOnTheOperatorsSharedSchedulerWhenCreatedWithAContext() { + var pollingThreadNames = new CopyOnWriteArrayList(); + when(resourceFetcher.fetchResources()) + .thenAnswer( + invocation -> { + pollingThreadNames.add(Thread.currentThread().getName()); + return testResponseWithOneValue(); + }); + + var configurationService = new BaseConfigurationService(); + var executorServiceManager = configurationService.getExecutorServiceManager(); + var eventSource = + new PollingEventSource( + SampleExternalResource.class, + contextFor(configurationService), + new PollingConfiguration<>(null, resourceFetcher, POLL_PERIOD, null)); + eventSource.setEventHandler(mock(EventHandler.class)); + + try { + eventSource.start(); + + // the initial fetch happens on the calling thread, the scheduled ones on the shared pool + await() + .untilAsserted( + () -> + assertThat(pollingThreadNames) + .anyMatch(name -> name.startsWith("josdk-scheduled-task-"))); + + eventSource.stop(); + + // the shared executor belongs to the operator, stopping an event source must not shut it down + assertThat(executorServiceManager.scheduledExecutorService().isShutdown()).isFalse(); + } finally { + executorServiceManager.stop(Duration.ofMillis(100)); + } + } + + @SuppressWarnings("unchecked") + private static EventSourceContext contextFor( + ConfigurationService configurationService) { + var controllerConfiguration = mock(ControllerConfiguration.class); + when(controllerConfiguration.getConfigurationService()).thenReturn(configurationService); + var context = mock(EventSourceContext.class); + when(context.getControllerConfiguration()).thenReturn(controllerConfiguration); + return context; + } + private Map> testResponseWithTwoValueForSameId() { Map> res = new HashMap<>(); res.put(primaryID1(), Set.of(testResource1(), testResource2())); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSourceTest.java index 3a4e1cb80d..3e6fbeaef3 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/timer/TimerEventSourceTest.java @@ -17,6 +17,7 @@ import java.util.List; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executors; import java.util.concurrent.TimeUnit; import org.awaitility.Awaitility; @@ -25,6 +26,7 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import io.javaoperatorsdk.operator.ExecutorUtilsInternal; import io.javaoperatorsdk.operator.TestUtils; import io.javaoperatorsdk.operator.api.reconciler.BaseControl; import io.javaoperatorsdk.operator.health.Status; @@ -125,6 +127,59 @@ public void handlesInstanceReschedule() { assertThat(eventHandler.events).hasSize(1); } + @Test + public void schedulesOnTheProvidedExecutorAndLeavesItRunningOnStop() { + var providedExecutor = + Executors.newSingleThreadScheduledExecutor( + ExecutorUtilsInternal.daemonThreadFactory("provided-timer-executor")); + var handler = new CapturingEventHandler(); + var eventSource = new TimerEventSource(() -> providedExecutor); + eventSource.setEventHandler(handler); + + try { + eventSource.start(); + eventSource.scheduleOnce(ResourceID.fromResource(TestUtils.testCustomResource()), PERIOD); + + untilAsserted( + () -> + assertThat(handler.eventProducingThreadNames) + .containsExactly("provided-timer-executor-1")); + + eventSource.stop(); + + // the executor is not the event source's to shut down + assertThat(providedExecutor.isShutdown()).isFalse(); + } finally { + providedExecutor.shutdownNow(); + } + } + + @Test + public void shutsDownTheExecutorItCreatedItselfAndCreatesANewOneOnRestart() { + var eventSource = new TimerEventSource(); + var handler = new CapturingEventHandler(); + eventSource.setEventHandler(handler); + + eventSource.start(); + eventSource.scheduleOnce(ResourceID.fromResource(TestUtils.testCustomResource()), PERIOD); + untilAsserted(() -> assertThat(handler.events).hasSize(1)); + var firstThreadName = handler.eventProducingThreadNames.get(0); + assertThat(firstThreadName).startsWith("josdk-timer-"); + + eventSource.stop(); + Awaitility.await() + .untilAsserted( + () -> + assertThat(Thread.getAllStackTraces().keySet()) + .noneMatch(t -> t.getName().equals(firstThreadName))); + + eventSource.start(); + eventSource.scheduleOnce(ResourceID.fromResource(TestUtils.testCustomResource()), PERIOD); + untilAsserted(() -> assertThat(handler.events).hasSize(2)); + + eventSource.stop(); + } + private void untilAsserted(ThrowingRunnable assertion) { untilAsserted(INITIAL_DELAY, PERIOD, assertion); } @@ -150,10 +205,12 @@ private void untilAsserted(long initialDelay, long interval, ThrowingRunnable as public static class CapturingEventHandler implements EventHandler { private final List events = new CopyOnWriteArrayList<>(); + private final List eventProducingThreadNames = new CopyOnWriteArrayList<>(); @Override public void handleEvent(Event event) { events.add(event); + eventProducingThreadNames.add(Thread.currentThread().getName()); } } } diff --git a/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java index 1a221fb76c..35e0bda882 100644 --- a/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java +++ b/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java @@ -82,10 +82,22 @@ public static ConfigLoader getDefault() { "workflow.executor-threads", Integer.class, ConfigurationServiceOverrider::withConcurrentWorkflowExecutorThreads), + new ConfigBinding<>( + "scheduled-tasks.concurrent-threads", + Integer.class, + ConfigurationServiceOverrider::withConcurrentScheduledTaskThreads), + new ConfigBinding<>( + "retry-and-reschedule.concurrent-threads", + Integer.class, + ConfigurationServiceOverrider::withConcurrentRetryAndRescheduleThreads), new ConfigBinding<>( "close-client-on-stop", Boolean.class, ConfigurationServiceOverrider::withCloseClientOnStop), + new ConfigBinding<>( + "use-virtual-threads", + Boolean.class, + ConfigurationServiceOverrider::withUseVirtualThreads), new ConfigBinding<>( "informer.stop-on-error-during-startup", Boolean.class, diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsCustomResource.java new file mode 100644 index 0000000000..be086a5f1b --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsCustomResource.java @@ -0,0 +1,30 @@ +/* + * 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.baseapi.virtualthreads; + +import io.fabric8.kubernetes.api.model.Namespaced; +import io.fabric8.kubernetes.client.CustomResource; +import io.fabric8.kubernetes.model.annotation.Group; +import io.fabric8.kubernetes.model.annotation.Kind; +import io.fabric8.kubernetes.model.annotation.ShortNames; +import io.fabric8.kubernetes.model.annotation.Version; + +@Group("sample.javaoperatorsdk") +@Version("v1") +@Kind("VirtualThreadsCustomResource") +@ShortNames("vtc") +public class VirtualThreadsCustomResource extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsIT.java new file mode 100644 index 0000000000..67ed2c780d --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsIT.java @@ -0,0 +1,92 @@ +/* + * 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.baseapi.virtualthreads; + +import java.util.concurrent.TimeUnit; +import java.util.stream.IntStream; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.ObjectMeta; +import io.javaoperatorsdk.annotation.Sample; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static io.javaoperatorsdk.operator.baseapi.virtualthreads.VirtualThreadsTestReconciler.onVirtualThread; +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +@Sample( + tldr = "Running reconciliations on virtual threads", + description = + """ + Demonstrates how to make the framework execute its concurrent work on virtual threads by \ + simply setting a flag on the ConfigurationService. Virtual threads make the blocking calls \ + a reconciler typically performs much cheaper, while the configured concurrency limits are \ + still enforced: the test verifies that the reconciler runs on virtual threads and that no \ + more than the configured number of reconciliations happen at the same time. + """) +class VirtualThreadsIT { + + static final int CONCURRENT_RECONCILIATION_THREADS = 2; + static final int NUMBER_OF_RESOURCES = 10; + + /** + * Virtual threads require Java 25, on an older JVM the framework transparently falls back to + * platform threads, which this test also covers since only the concurrency assertions apply then. + */ + private static final boolean VIRTUAL_THREADS_SUPPORTED = Runtime.version().feature() >= 25; + + @RegisterExtension + LocallyRunOperatorExtension operator = + LocallyRunOperatorExtension.builder() + .withConfigurationService( + o -> + o.withUseVirtualThreads(true) + .withConcurrentReconciliationThreads(CONCURRENT_RECONCILIATION_THREADS)) + .withReconciler(new VirtualThreadsTestReconciler()) + .build(); + + @Test + void reconciliationsRunOnVirtualThreadsWithinTheConfiguredConcurrency() { + // the test itself runs on a platform thread, so the reconciler assertion below can only pass + // if the framework actually switched to virtual threads + assertThat(onVirtualThread()).isFalse(); + + IntStream.range(0, NUMBER_OF_RESOURCES).forEach(i -> operator.create(testResource(i))); + + var reconciler = operator.getReconcilerOfType(VirtualThreadsTestReconciler.class); + await() + .atMost(2, TimeUnit.MINUTES) + .untilAsserted( + () -> + assertThat(reconciler.getNumberOfExecutions()) + .isGreaterThanOrEqualTo(NUMBER_OF_RESOURCES)); + + // parallelism is retained: reconciliations do happen concurrently, but never more than the + // configured number of them + assertThat(reconciler.getMaxConcurrentReconciliations()) + .isEqualTo(CONCURRENT_RECONCILIATION_THREADS); + assertThat(reconciler.allExecutionsOnVirtualThreads()).isEqualTo(VIRTUAL_THREADS_SUPPORTED); + } + + private VirtualThreadsCustomResource testResource(int index) { + var resource = new VirtualThreadsCustomResource(); + resource.setMetadata(new ObjectMeta()); + resource.getMetadata().setName("virtual-threads-test-" + index); + return resource; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsTestReconciler.java new file mode 100644 index 0000000000..ee6f00f499 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/virtualthreads/VirtualThreadsTestReconciler.java @@ -0,0 +1,84 @@ +/* + * 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.baseapi.virtualthreads; + +import java.time.Duration; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; + +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider; + +/** + * Blocks for a while during reconciliation, recording on what kind of thread it ran and how many + * reconciliations were in flight at the same time. + */ +@ControllerConfiguration +public class VirtualThreadsTestReconciler + implements Reconciler, TestExecutionInfoProvider { + + public static final Duration RECONCILIATION_DURATION = Duration.ofMillis(300); + + private final AtomicInteger numberOfExecutions = new AtomicInteger(); + private final AtomicInteger runningReconciliations = new AtomicInteger(); + private final AtomicInteger maxConcurrentReconciliations = new AtomicInteger(); + private final AtomicBoolean allExecutionsOnVirtualThreads = new AtomicBoolean(true); + + @Override + public UpdateControl reconcile( + VirtualThreadsCustomResource resource, Context context) + throws InterruptedException { + if (!onVirtualThread()) { + allExecutionsOnVirtualThreads.set(false); + } + maxConcurrentReconciliations.accumulateAndGet( + runningReconciliations.incrementAndGet(), Math::max); + try { + Thread.sleep(RECONCILIATION_DURATION.toMillis()); + } finally { + runningReconciliations.decrementAndGet(); + numberOfExecutions.incrementAndGet(); + } + return UpdateControl.noUpdate(); + } + + public int getNumberOfExecutions() { + return numberOfExecutions.get(); + } + + public int getMaxConcurrentReconciliations() { + return maxConcurrentReconciliations.get(); + } + + public boolean allExecutionsOnVirtualThreads() { + return allExecutionsOnVirtualThreads.get(); + } + + /** + * {@code Thread.isVirtual} only exists as of Java 21 while the tests are compiled for Java 17, + * hence the reflective call. + */ + static boolean onVirtualThread() { + try { + return (Boolean) Thread.class.getMethod("isVirtual").invoke(Thread.currentThread()); + } catch (ReflectiveOperationException e) { + return false; + } + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/loader/ConfigLoaderTest.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/loader/ConfigLoaderTest.java index 0b014bb6f7..13f44b036a 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/loader/ConfigLoaderTest.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/config/loader/ConfigLoaderTest.java @@ -86,6 +86,30 @@ void applyConfigsAppliesConcurrentWorkflowExecutorThreads() { assertThat(result.concurrentWorkflowExecutorThreads()).isEqualTo(3); } + @Test + void applyConfigsAppliesConcurrentScheduledTaskThreads() { + var loader = + new ConfigLoader(mapProvider(Map.of("josdk.scheduled-tasks.concurrent-threads", 8))); + + var base = new BaseConfigurationService(null); + var result = + ConfigurationService.newOverriddenConfigurationService(base, loader.applyConfigs()); + + assertThat(result.concurrentScheduledTaskThreads()).isEqualTo(8); + } + + @Test + void applyConfigsAppliesConcurrentRetryAndRescheduleThreads() { + var loader = + new ConfigLoader(mapProvider(Map.of("josdk.retry-and-reschedule.concurrent-threads", 6))); + + var base = new BaseConfigurationService(null); + var result = + ConfigurationService.newOverriddenConfigurationService(base, loader.applyConfigs()); + + assertThat(result.concurrentRetryAndRescheduleThreads()).isEqualTo(6); + } + @Test void applyConfigsAppliesBooleanFlags() { var values = new HashMap(); @@ -95,6 +119,7 @@ void applyConfigsAppliesBooleanFlags() { values.put("josdk.dependent-resources.ssa-based-create-update-match", false); values.put("josdk.use-ssa-to-patch-primary-resource", false); values.put("josdk.clone-secondary-resources-when-getting-from-cache", true); + values.put("josdk.use-virtual-threads", true); var loader = new ConfigLoader(mapProvider(values)); var base = new BaseConfigurationService(null); @@ -107,6 +132,7 @@ void applyConfigsAppliesBooleanFlags() { assertThat(result.ssaBasedCreateUpdateMatchForDependentResources()).isFalse(); assertThat(result.useSSAToPatchPrimaryResource()).isFalse(); assertThat(result.cloneSecondaryResourcesWhenGettingFromCache()).isTrue(); + assertThat(result.useVirtualThreads()).isTrue(); } @Test @@ -311,7 +337,11 @@ public Optional getValue(String key, Class type) { Map.entry( "withCloneSecondaryResourcesWhenGettingFromCache", "clone-secondary-resources-when-getting-from-cache"), - Map.entry("withClusterScopedEventNamespace", "events.cluster-scoped-namespace")); + Map.entry("withClusterScopedEventNamespace", "events.cluster-scoped-namespace"), + Map.entry( + "withConcurrentRetryAndRescheduleThreads", "retry-and-reschedule.concurrent-threads"), + Map.entry("withConcurrentScheduledTaskThreads", "scheduled-tasks.concurrent-threads"), + Map.entry("withUseVirtualThreads", "use-virtual-threads")); /** * Maps every scalar setter of {@link ControllerConfigurationOverrider} to the property key suffix diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/multiplemanagedexternaldependenttype/MultipleManagedExternalDependentResourceReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/multiplemanagedexternaldependenttype/MultipleManagedExternalDependentResourceReconciler.java index dec9422486..33b7f965f3 100644 --- a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/multiplemanagedexternaldependenttype/MultipleManagedExternalDependentResourceReconciler.java +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/multiplemanagedexternaldependenttype/MultipleManagedExternalDependentResourceReconciler.java @@ -96,6 +96,7 @@ public int getNumberOfExecutions() { pollingEventSource = new PollingEventSource<>( ExternalResource.class, + context, new PollingConfigurationBuilder( fetcher, Duration.ofMillis(1000L)) .withName(EVENT_SOURCE_NAME) diff --git a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageOperator.java b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageOperator.java index 5366dc2e9a..392172886f 100644 --- a/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageOperator.java +++ b/sample-operators/webpage/src/main/java/io/javaoperatorsdk/operator/sample/WebPageOperator.java @@ -40,7 +40,9 @@ public class WebPageOperator { public static void main(String[] args) throws IOException { log.info("WebServer Operator starting!"); - Operator operator = new Operator(o -> o.withStopOnInformerErrorDuringStartup(false)); + Operator operator = + new Operator( + o -> o.withStopOnInformerErrorDuringStartup(false).withUseVirtualThreads(true)); String reconcilerEnvVar = System.getenv(WEBPAGE_RECONCILER_ENV); if (WEBPAGE_CLASSIC_RECONCILER_ENV_VALUE.equals(reconcilerEnvVar)) { operator.register(new WebPageReconciler()); diff --git a/spotbugs-exclude.xml b/spotbugs-exclude.xml index f6107ac4ea..9c516e059e 100644 --- a/spotbugs-exclude.xml +++ b/spotbugs-exclude.xml @@ -67,4 +67,14 @@ + + + + + +