diff --git a/.github/workflows/e2e-test.yml b/.github/workflows/e2e-test.yml index 6a86b16f69..530784eb83 100644 --- a/.github/workflows/e2e-test.yml +++ b/.github/workflows/e2e-test.yml @@ -26,6 +26,8 @@ jobs: - "sample-operators/leader-election" - "sample-operators/operations" - "sample-operators/kotlin-operator" + # requires Java 21+, only built with such a JDK, see sample-operators/pom.xml + - "sample-operators/virtual-threads" runs-on: ubuntu-latest steps: - name: Checkout diff --git a/docs/content/en/docs/documentation/operations/configuration.md b/docs/content/en/docs/documentation/operations/configuration.md index 513cc432d8..88b33aef17 100644 --- a/docs/content/en/docs/documentation/operations/configuration.md +++ b/docs/content/en/docs/documentation/operations/configuration.md @@ -23,6 +23,47 @@ Operator operator = new Operator( override -> override .withLeaderElectionConfiguration(new LeaderElectionConfiguration("bar", "barNS"))); ``` +## Running on Virtual Threads + +Reconciliations and workflow steps are executed on the executors provided by +`ConfigurationService`, which are fixed size thread pools by default. On Java 21 and newer, these +can be replaced with virtual threads, so that a reconciliation waiting on a remote call does not +occupy a platform thread: + +```java +Operator operator = new Operator(override -> override + .withExecutorService(Executors.newVirtualThreadPerTaskExecutor()) + .withWorkflowExecutorService(Executors.newVirtualThreadPerTaskExecutor())); +``` + +Since a virtual thread per task executor is not a pool, `concurrentReconciliationThreads` and +`concurrentWorkflowExecutorThreads` have no effect in this case: the number of parallel +reconciliations is not limited anymore. Reconciliations of the same resource are still serialized +by the framework. + +The Kubernetes client runs its asynchronous tasks - most notably the informer event handlers the +framework registers - on its own task executor, which can be configured similarly: + +```java +KubernetesClient client = new KubernetesClientBuilder() + .withTaskExecutorSupplier(new KubernetesClientBuilder.ExecutorSupplier() { + @Override + public Executor get() { + return Executors.newVirtualThreadPerTaskExecutor(); + } + + @Override + public void onClose(Executor executor) { + ((ExecutorService) executor).shutdownNow(); + } + }) + .build(); +``` + +See the +[virtual threads sample](https://github.com/operator-framework/java-operator-sdk/tree/main/sample-operators/virtual-threads) +for a complete operator using both. + ## 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: diff --git a/docs/content/en/docs/getting-started/bootstrap-and-samples.md b/docs/content/en/docs/getting-started/bootstrap-and-samples.md index d0d94f860d..ccc39c7609 100644 --- a/docs/content/en/docs/getting-started/bootstrap-and-samples.md +++ b/docs/content/en/docs/getting-started/bootstrap-and-samples.md @@ -50,6 +50,11 @@ The [sample-operators](https://github.com/java-operator-sdk/java-operator-sdk/tr - **Key Features**: Multiple controllers managing related custom resources - **Good for**: Understanding complex operators with multiple resource types and relationships +**[virtual-threads](https://github.com/operator-framework/java-operator-sdk/tree/main/sample-operators/virtual-threads)** +- **Purpose**: Runs reconciliations and Kubernetes client callbacks on virtual threads +- **Key Features**: Requires Java 21 or newer, configures both the framework and the client executors +- **Good for**: Operators with blocking reconciliations, see also [configuration](../../documentation/operations/configuration#running-on-virtual-threads) + ## Running the Samples ### Prerequisites diff --git a/sample-operators/pom.xml b/sample-operators/pom.xml index 704007c076..de49229ac5 100644 --- a/sample-operators/pom.xml +++ b/sample-operators/pom.xml @@ -38,4 +38,17 @@ operations kotlin-operator + + + + + virtual-threads-sample + + [21,) + + + virtual-threads + + + diff --git a/sample-operators/virtual-threads/README.md b/sample-operators/virtual-threads/README.md new file mode 100644 index 0000000000..a5a2c4b6a5 --- /dev/null +++ b/sample-operators/virtual-threads/README.md @@ -0,0 +1,66 @@ +# Virtual Threads Sample + +This sample runs an operator on [Java virtual threads](https://openjdk.org/jeps/444). +**It requires Java 21 or newer**, therefore the module is only part of the build when the JDK +used is at least version 21 (see the profile in the parent `sample-operators/pom.xml`). + +Two independent places execute the code of an operator, both are configured in +[`VirtualThreads`](src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreads.java): + +1. The framework's executors, used to run reconciliations and workflow steps: + + ```java + new Operator(overrider -> overrider + .withExecutorService(Executors.newVirtualThreadPerTaskExecutor()) + .withWorkflowExecutorService(Executors.newVirtualThreadPerTaskExecutor())); + ``` + +2. The Kubernetes client's task executor, used for the asynchronous tasks of the client, most + notably the informer event handlers registered by the framework: + + ```java + new KubernetesClientBuilder() + .withTaskExecutorSupplier(new KubernetesClientBuilder.ExecutorSupplier() { + @Override + public Executor get() { + return Executors.newVirtualThreadPerTaskExecutor(); + } + + @Override + public void onClose(Executor executor) { + ((ExecutorService) executor).shutdownNow(); + } + }) + .build(); + ``` + +Note that a virtual thread per task executor is not a pool, therefore +`concurrentReconciliationThreads` and `concurrentWorkflowExecutorThreads` have no effect: the +number of parallel reconciliations is not limited anymore. Reconciliations of the same resource +are still serialized by the framework, as with the default executors. + +The [reconciler](src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsReconciler.java) +of this sample simulates a slow remote call by sleeping for a second, and records in the status +whether it was reconciled on a virtual thread. Since a virtual thread does not occupy a platform +thread while it waits, an arbitrary number of such reconciliations can run in parallel. + +## Running the Sample + +```shell +kubectl apply -f target/classes/META-INF/fabric8/virtualthreadscustomresources.sample.javaoperatorsdk-v1.yml +mvn exec:java -Dexec.mainClass="io.javaoperatorsdk.operator.sample.VirtualThreadsOperator" +``` + +Then create a custom resource: + +```shell +kubectl apply -f k8s/virtual-threads-custom-resource.yaml +``` + +The status of the resource shows that the reconciliation happened on a virtual thread: + +```yaml +status: + observedValue: "initial value" + reconciledOnVirtualThread: true +``` diff --git a/sample-operators/virtual-threads/k8s/operator.yaml b/sample-operators/virtual-threads/k8s/operator.yaml new file mode 100644 index 0000000000..9ca1707b56 --- /dev/null +++ b/sample-operators/virtual-threads/k8s/operator.yaml @@ -0,0 +1,75 @@ +# +# 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. +# + +apiVersion: v1 +kind: ServiceAccount +metadata: + name: virtual-threads-operator + +--- +apiVersion: apps/v1 +kind: Deployment +metadata: + name: virtual-threads-operator +spec: + selector: + matchLabels: + app: virtual-threads-operator + replicas: 1 + template: + metadata: + labels: + app: virtual-threads-operator + spec: + serviceAccountName: virtual-threads-operator + containers: + - name: operator + image: virtual-threads-operator + imagePullPolicy: Never + +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: virtual-threads-operator-admin +subjects: +- kind: ServiceAccount + name: virtual-threads-operator + namespace: default +roleRef: + kind: ClusterRole + name: virtual-threads-operator + apiGroup: "" + +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: virtual-threads-operator +rules: +- apiGroups: + - "apiextensions.k8s.io" + resources: + - customresourcedefinitions + verbs: + - '*' +- apiGroups: + - "sample.javaoperatorsdk" + resources: + - virtualthreadscustomresources + - virtualthreadscustomresources/status + verbs: + - '*' diff --git a/sample-operators/virtual-threads/k8s/virtual-threads-custom-resource.yaml b/sample-operators/virtual-threads/k8s/virtual-threads-custom-resource.yaml new file mode 100644 index 0000000000..124ecb4058 --- /dev/null +++ b/sample-operators/virtual-threads/k8s/virtual-threads-custom-resource.yaml @@ -0,0 +1,22 @@ +# +# 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. +# + +apiVersion: "sample.javaoperatorsdk/v1" +kind: VirtualThreadsCustomResource +metadata: + name: test1 +spec: + value: "initial value" diff --git a/sample-operators/virtual-threads/pom.xml b/sample-operators/virtual-threads/pom.xml new file mode 100644 index 0000000000..2fbb255f43 --- /dev/null +++ b/sample-operators/virtual-threads/pom.xml @@ -0,0 +1,111 @@ + + + + 4.0.0 + + + io.javaoperatorsdk + sample-operators + 999-SNAPSHOT + + + sample-virtual-threads-operator + jar + Operator SDK - Samples - Virtual Threads + Runs reconciliations and Kubernetes client callbacks on virtual threads, requires Java 21+ + + + + 21 + + + + + + io.javaoperatorsdk + operator-framework-bom + ${project.version} + pom + import + + + + + + + io.javaoperatorsdk + operator-framework + + + org.apache.logging.log4j + log4j-slf4j2-impl + compile + + + org.apache.logging.log4j + log4j-core + compile + + + org.awaitility + awaitility + test + + + io.javaoperatorsdk + operator-framework-junit + test + + + + + + com.google.cloud.tools + jib-maven-plugin + ${jib-maven-plugin.version} + + + gcr.io/distroless/java21-debian12 + + + virtual-threads-operator + + + + + org.apache.maven.plugins + maven-compiler-plugin + + + io.fabric8 + crd-generator-maven-plugin + ${fabric8-client.version} + + + + generate + + + + + + + + diff --git a/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreads.java b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreads.java new file mode 100644 index 0000000000..e43cef6f4e --- /dev/null +++ b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreads.java @@ -0,0 +1,71 @@ +/* + * 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.sample; + +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; + +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientBuilder; +import io.javaoperatorsdk.operator.api.config.ConfigurationServiceOverrider; + +/** + * Wiring required to run an operator on virtual threads. Used both by {@link + * VirtualThreadsOperator} and by the tests of this sample, so that the operator behaves the same + * way when it runs locally and when it runs inside a cluster. + */ +public final class VirtualThreads { + + private VirtualThreads() {} + + /** + * The Kubernetes client executes its asynchronous tasks - most notably the informer event + * handlers the framework registers - on the task executor. Backing it with virtual threads means + * that blocking such a callback no longer blocks a platform thread. + */ + public static KubernetesClient newKubernetesClient() { + return new KubernetesClientBuilder() + .withTaskExecutorSupplier(new VirtualThreadExecutorSupplier()) + .build(); + } + + /** + * Reconciliations and workflow steps are executed on the executors configured here. Note that + * {@code concurrentReconciliationThreads} and {@code concurrentWorkflowExecutorThreads} have no + * effect anymore, since a virtual thread per task executor is not pooled and therefore unbounded. + * Reconciliations of the same resource are still serialized by the framework. + */ + public static void configureExecutors(ConfigurationServiceOverrider overrider) { + overrider + .withExecutorService(Executors.newVirtualThreadPerTaskExecutor()) + .withWorkflowExecutorService(Executors.newVirtualThreadPerTaskExecutor()); + } + + private static class VirtualThreadExecutorSupplier + implements KubernetesClientBuilder.ExecutorSupplier { + + @Override + public Executor get() { + return Executors.newVirtualThreadPerTaskExecutor(); + } + + @Override + public void onClose(Executor executor) { + ((ExecutorService) executor).shutdownNow(); + } + } +} diff --git a/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsCustomResource.java b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsCustomResource.java new file mode 100644 index 0000000000..93a75f974e --- /dev/null +++ b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsCustomResource.java @@ -0,0 +1,26 @@ +/* + * 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.sample; + +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.Version; + +@Group("sample.javaoperatorsdk") +@Version("v1") +public class VirtualThreadsCustomResource + extends CustomResource implements Namespaced {} diff --git a/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsOperator.java b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsOperator.java new file mode 100644 index 0000000000..2ce51eba9b --- /dev/null +++ b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsOperator.java @@ -0,0 +1,40 @@ +/* + * 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.sample; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.javaoperatorsdk.operator.Operator; + +public class VirtualThreadsOperator { + + private static final Logger log = LoggerFactory.getLogger(VirtualThreadsOperator.class); + + public static void main(String[] args) { + log.info("Virtual Threads Operator starting!"); + + Operator operator = + new Operator( + overrider -> { + overrider.withKubernetesClient(VirtualThreads.newKubernetesClient()); + VirtualThreads.configureExecutors(overrider); + }); + operator.register(new VirtualThreadsReconciler()); + operator.installShutdownHook(); + operator.start(); + } +} diff --git a/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsReconciler.java b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsReconciler.java new file mode 100644 index 0000000000..a8c3718c5d --- /dev/null +++ b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsReconciler.java @@ -0,0 +1,78 @@ +/* + * 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.sample; + +import java.time.Duration; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; + +/** + * Simulates a reconciliation that spends most of its time waiting on a remote call. On virtual + * threads such a wait does not hold on to a platform thread, thus an arbitrary number of resources + * can be reconciled in parallel. + */ +public class VirtualThreadsReconciler implements Reconciler { + + public static final Duration BLOCKING_CALL_DURATION = Duration.ofSeconds(1); + + private static final Logger log = LoggerFactory.getLogger(VirtualThreadsReconciler.class); + + @Override + public UpdateControl reconcile( + VirtualThreadsCustomResource resource, Context context) { + var thread = Thread.currentThread(); + log.info( + "Reconciling: {} on thread: {}, virtual: {}", + resource.getMetadata().getName(), + thread.getName(), + thread.isVirtual()); + + simulateBlockingCall(); + + var response = createResponseResource(resource); + response.getStatus().setObservedValue(resource.getSpec().getValue()); + response.getStatus().setReconciledOnVirtualThread(thread.isVirtual()); + + return UpdateControl.patchStatus(response); + } + + private VirtualThreadsCustomResource createResponseResource( + VirtualThreadsCustomResource resource) { + var res = new VirtualThreadsCustomResource(); + res.setMetadata( + new ObjectMetaBuilder() + .withName(resource.getMetadata().getName()) + .withNamespace(resource.getMetadata().getNamespace()) + .build()); + res.setStatus(new VirtualThreadsStatus()); + return res; + } + + private void simulateBlockingCall() { + try { + Thread.sleep(BLOCKING_CALL_DURATION); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new IllegalStateException(e); + } + } +} diff --git a/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsSpec.java b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsSpec.java new file mode 100644 index 0000000000..fe590f907f --- /dev/null +++ b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsSpec.java @@ -0,0 +1,29 @@ +/* + * 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.sample; + +public class VirtualThreadsSpec { + + private String value; + + public String getValue() { + return value; + } + + public void setValue(String value) { + this.value = value; + } +} diff --git a/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsStatus.java b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsStatus.java new file mode 100644 index 0000000000..93d53e2cee --- /dev/null +++ b/sample-operators/virtual-threads/src/main/java/io/javaoperatorsdk/operator/sample/VirtualThreadsStatus.java @@ -0,0 +1,38 @@ +/* + * 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.sample; + +public class VirtualThreadsStatus { + + private String observedValue; + private Boolean reconciledOnVirtualThread; + + public String getObservedValue() { + return observedValue; + } + + public void setObservedValue(String observedValue) { + this.observedValue = observedValue; + } + + public Boolean getReconciledOnVirtualThread() { + return reconciledOnVirtualThread; + } + + public void setReconciledOnVirtualThread(Boolean reconciledOnVirtualThread) { + this.reconciledOnVirtualThread = reconciledOnVirtualThread; + } +} diff --git a/sample-operators/virtual-threads/src/main/resources/log4j2.xml b/sample-operators/virtual-threads/src/main/resources/log4j2.xml new file mode 100644 index 0000000000..147f494c1d --- /dev/null +++ b/sample-operators/virtual-threads/src/main/resources/log4j2.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + diff --git a/sample-operators/virtual-threads/src/test/java/io/javaoperatorsdk/operator/sample/VirtualThreadsOperatorE2E.java b/sample-operators/virtual-threads/src/test/java/io/javaoperatorsdk/operator/sample/VirtualThreadsOperatorE2E.java new file mode 100644 index 0000000000..e27c985d4c --- /dev/null +++ b/sample-operators/virtual-threads/src/test/java/io/javaoperatorsdk/operator/sample/VirtualThreadsOperatorE2E.java @@ -0,0 +1,128 @@ +/* + * 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.sample; + +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.time.Duration; +import java.util.List; +import java.util.stream.IntStream; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientBuilder; +import io.javaoperatorsdk.operator.junit.AbstractOperatorExtension; +import io.javaoperatorsdk.operator.junit.ClusterDeployedOperatorExtension; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +class VirtualThreadsOperatorE2E { + + static final Logger log = LoggerFactory.getLogger(VirtualThreadsOperatorE2E.class); + + static final KubernetesClient client = new KubernetesClientBuilder().build(); + + public static final int RESOURCE_COUNT = 20; + public static final String INITIAL_VALUE = "initial value"; + public static final String CHANGED_VALUE = "changed value"; + public static final Duration WAIT_TIMEOUT = Duration.ofSeconds(120); + + @RegisterExtension + AbstractOperatorExtension operator = + isLocal() + ? LocallyRunOperatorExtension.builder() + .withKubernetesClient(VirtualThreads.newKubernetesClient()) + .withConfigurationService(VirtualThreads::configureExecutors) + .withReconciler(new VirtualThreadsReconciler()) + .build() + : ClusterDeployedOperatorExtension.builder() + .withOperatorDeployment(client.load(new FileInputStream("k8s/operator.yaml")).items()) + .build(); + + public VirtualThreadsOperatorE2E() throws FileNotFoundException {} + + /** + * All the resources block during their reconciliation, still all of them are reconciled in + * parallel, without occupying a platform thread while waiting. + */ + @Test + void reconcilesAllResourcesOnVirtualThreads() { + testResources(INITIAL_VALUE).forEach(r -> operator.create(r)); + + awaitObservedValue(INITIAL_VALUE); + + testResources(CHANGED_VALUE).forEach(r -> operator.replace(r)); + + awaitObservedValue(CHANGED_VALUE); + + testResources(CHANGED_VALUE).forEach(r -> operator.delete(r)); + + await() + .atMost(WAIT_TIMEOUT) + .untilAsserted( + () -> + assertThat(operator.resources(VirtualThreadsCustomResource.class).list().getItems()) + .isEmpty()); + } + + void awaitObservedValue(String value) { + await() + .atMost(WAIT_TIMEOUT) + .untilAsserted( + () -> { + for (int i = 0; i < RESOURCE_COUNT; i++) { + var actual = operator.get(VirtualThreadsCustomResource.class, resourceName(i)); + assertThat(actual.getStatus()).isNotNull(); + assertThat(actual.getStatus().getObservedValue()).isEqualTo(value); + assertThat(actual.getStatus().getReconciledOnVirtualThread()).isTrue(); + } + }); + } + + List testResources(String value) { + return IntStream.range(0, RESOURCE_COUNT).mapToObj(i -> testResource(i, value)).toList(); + } + + VirtualThreadsCustomResource testResource(int index, String value) { + var resource = new VirtualThreadsCustomResource(); + resource.setMetadata( + new ObjectMetaBuilder() + .withName(resourceName(index)) + .withNamespace(operator.getNamespace()) + .build()); + resource.setSpec(new VirtualThreadsSpec()); + resource.getSpec().setValue(value); + return resource; + } + + String resourceName(int index) { + return "test-" + index; + } + + boolean isLocal() { + var deployment = System.getProperty("test.deployment"); + boolean remote = deployment != null && deployment.equals("remote"); + log.info("Running the operator {}", remote ? "remote" : "locally"); + return !remote; + } +} diff --git a/sample-operators/virtual-threads/src/test/resources/log4j2.xml b/sample-operators/virtual-threads/src/test/resources/log4j2.xml new file mode 100644 index 0000000000..8b1c5ca270 --- /dev/null +++ b/sample-operators/virtual-threads/src/test/resources/log4j2.xml @@ -0,0 +1,30 @@ + + + + + + + + + + + + + + \ No newline at end of file