diff --git a/docs/content/en/docs/documentation/operations/configuration.md b/docs/content/en/docs/documentation/operations/configuration.md index cdfb1b7fdb..45849a5ceb 100644 --- a/docs/content/en/docs/documentation/operations/configuration.md +++ b/docs/content/en/docs/documentation/operations/configuration.md @@ -23,6 +23,34 @@ 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. + +Two things to keep in mind: + +- Virtual threads require Java 21 or later at runtime. When the flag is set on an older JVM, a + warning is logged and platform threads are used instead, so the same configuration works on any + supported Java version. +- A custom `ExecutorService` provided through `withExecutorService(...)` or + `withWorkflowExecutorService(...)` is always used as is, the flag has no effect on it. + ## 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: @@ -265,6 +293,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 (requires Java 21+ at runtime) | | `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 | 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 35f46e5019..e3454e6c98 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,6 @@ import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.function.Consumer; import org.slf4j.Logger; @@ -228,6 +227,34 @@ 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. + * + *

Requires Java 21 or later at runtime. When enabled 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()} is used as is. + * + * @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 +263,8 @@ default Metrics getMetrics() { * processing */ default ExecutorService getExecutorService() { - return Executors.newFixedThreadPool(concurrentReconciliationThreads()); + return ExecutorServiceManager.newBoundedExecutorService( + concurrentReconciliationThreads(), useVirtualThreads()); } /** @@ -246,7 +274,8 @@ 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()); } /** 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..b3ae079561 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 @@ -50,6 +50,7 @@ public class ConfigurationServiceOverrider { private KubernetesClient client; private ExecutorService executorService; private ExecutorService workflowExecutorService; + private Boolean useVirtualThreads; private LeaderElectionConfiguration leaderElectionConfiguration; private String clusterScopedEventNamespace; private EventRecorder eventRecorder; @@ -119,6 +120,19 @@ public ConfigurationServiceOverrider withWorkflowExecutorService( return this; } + /** + * Makes the framework run the tasks it executes concurrently on virtual threads instead of + * platform threads. Requires Java 21 or later at runtime, 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. @@ -322,6 +336,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) { 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..2176cb0fab 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 @@ -49,6 +49,37 @@ public class ExecutorServiceManager { 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 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) @@ -135,7 +166,8 @@ public ScheduledExecutorService scheduledExecutorService() { public synchronized void start(ConfigurationService configurationService) { if (!started) { this.configurationService = configurationService; // used to lazy init workflow executor - this.cachingExecutorService = Executors.newCachedThreadPool(); + this.cachingExecutorService = + newUnboundedExecutorService(configurationService.useVirtualThreads()); this.scheduledExecutorService = Executors.newScheduledThreadPool(0); this.executor = new InstrumentedExecutorService(configurationService.getExecutorService()); started = true; 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..dd7a8a80a0 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java @@ -0,0 +1,178 @@ +/* + * 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.List; +import java.util.concurrent.AbstractExecutorService; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +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.javaoperatorsdk.operator.OperatorException; + +/** + * Creates the virtual thread based executors used when {@link + * ConfigurationService#useVirtualThreads()} is enabled. + * + *

The SDK is compiled for Java 17, in which virtual threads don't exist yet, so {@code + * Executors.newVirtualThreadPerTaskExecutor()} is looked up reflectively and is only available when + * the operator actually runs on Java 21 or later. + */ +final class VirtualThreads { + + private static final Logger log = LoggerFactory.getLogger(VirtualThreads.class); + + 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 21 or later. */ + static boolean isSupported() { + return NEW_VIRTUAL_THREAD_PER_TASK_EXECUTOR != null; + } + + /** + * 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 not supported by the JVM in use (Java {}, Java 21" + + " or later is required). Falling back to platform threads.", + 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 not supported by the JVM in use, Java 21 or later is required"); + } + 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); + } + } + + /** + * 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) { + 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 BoundedExecutorService(ExecutorService delegate, int maxConcurrency) { + this.delegate = delegate; + // fair, so that tasks run roughly in submission order as they would on a thread pool + this.permits = new Semaphore(maxConcurrency, true); + } + + @Override + public void execute(Runnable command) { + delegate.execute( + () -> { + try { + permits.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + // shutdownNow interrupted us before the task even started: cancel it so that whoever + // waits on the associated future isn't left hanging + if (command instanceof Future) { + ((Future) command).cancel(false); + } + return; + } + try { + command.run(); + } finally { + permits.release(); + } + }); + } + + @Override + public void shutdown() { + delegate.shutdown(); + } + + @Override + public List shutdownNow() { + return delegate.shutdownNow(); + } + + @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/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..0247e677c6 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 @@ -152,6 +152,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/VirtualThreadsTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java new file mode 100644 index 0000000000..60d7e26f54 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java @@ -0,0 +1,162 @@ +/* + * 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.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 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 + @EnabledForJreRange(min = JRE.JAVA_21) + 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_21) + 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_21) + 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_21) + 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(); + } + + /** + * {@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/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java b/operator-framework/src/main/java/io/javaoperatorsdk/operator/config/loader/ConfigLoader.java index c8daf89724..73f1893c72 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,6 +82,10 @@ public static ConfigLoader getDefault() { "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..fbb38042aa --- /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 21, 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() >= 21; + + @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 44fac32b7d..ec0f18981d 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 @@ -93,6 +93,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); @@ -105,6 +106,7 @@ void applyConfigsAppliesBooleanFlags() { assertThat(result.ssaBasedCreateUpdateMatchForDependentResources()).isFalse(); assertThat(result.useSSAToPatchPrimaryResource()).isFalse(); assertThat(result.cloneSecondaryResourcesWhenGettingFromCache()).isTrue(); + assertThat(result.useVirtualThreads()).isTrue(); } @Test