diff --git a/bootstrapper-maven-plugin/pom.xml b/bootstrapper-maven-plugin/pom.xml index 75c1ebd50d..bcd6c2e743 100644 --- a/bootstrapper-maven-plugin/pom.xml +++ b/bootstrapper-maven-plugin/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk java-operator-sdk - 5.6.1-SNAPSHOT + 999-SNAPSHOT bootstrapper diff --git a/caffeine-bounded-cache-support/pom.xml b/caffeine-bounded-cache-support/pom.xml index 8ae3911352..be70ab9a2e 100644 --- a/caffeine-bounded-cache-support/pom.xml +++ b/caffeine-bounded-cache-support/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.6.1-SNAPSHOT + 999-SNAPSHOT caffeine-bounded-cache-support diff --git a/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md b/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md index 538c52c00c..3ac1e88e71 100644 --- a/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md +++ b/docs/content/en/docs/documentation/dependent-resource-and-workflows/dependent-resources.md @@ -531,6 +531,34 @@ samples [here](https://github.com/java-operator-sdk/java-operator-sdk/tree/main/ in [related integration test](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework/src/test/java/io/javaoperatorsdk/operator/workflow/orderedmanageddependent/ConfigMapDependentResource2.java) . +### Adding Common Metadata to All Managed Resources (Desired State Aspects) + +Operators often need to mark every resource they manage in a uniform way, for example with a +`app.kubernetes.io/managed-by` label, so that these resources can easily be identified, selected or +garbage-collected later on. Instead of repeating that logic in every `desired()` implementation, a +`DesiredStateAspect` can be registered once, at the operator level, and is then applied to the +desired state of every Kubernetes dependent resource managed by the operator: + +```java +Operator operator = new Operator(overrider -> overrider + .withDesiredStateAspects(List.of( + (desired, dependentResource, context) -> desired.getMetadata().getLabels() + .put("app.kubernetes.io/managed-by", "my-operator")))); +``` + +Aspects are applied, in registration order, right after the desired state has been computed and +before the desired state is matched against the actual resource, created or updated. As a +consequence, the metadata added by an aspect is part of the desired state proper: if it is removed +from the actual resource, or if the aspect itself changes, the associated secondary resources are +updated accordingly on the next reconciliation. + +Since the desired state is computed at most once per reconciliation and cached in the `Context`, +aspects are called at most once per dependent resource and reconciliation. They are only called for +dependent resources whose desired state is a `HasMetadata`, meaning that external (non-Kubernetes) +dependent resources are left untouched. Implementations are expected to modify the provided desired +state in place and need to be thread-safe as they can be called concurrently for different primary +resources. + ## "Read-only" Dependent Resources vs. Event Source See Integration test for a read-only diff --git a/docs/content/en/docs/documentation/eventing.md b/docs/content/en/docs/documentation/eventing.md index e7aea6b065..4b890407e4 100644 --- a/docs/content/en/docs/documentation/eventing.md +++ b/docs/content/en/docs/documentation/eventing.md @@ -348,6 +348,30 @@ See also [CaffeineBoundedItemStores](https://github.com/operator-framework/java-operator-sdk/blob/main/caffeine-bounded-cache-support/src/main/java/io/javaoperatorsdk/operator/processing/event/source/cache/CaffeineBoundedItemStores.java) for more details. +### Removing the Namespace Index + +Informers keep an index from namespace to the resources cached for it. JOSDK never reads that index, +only the ones registered explicitly through `IndexerResourceCache.addIndexers(..)`, so it can be +removed to save an entry per cached resource: + +```java +@ControllerConfiguration(informer = @Informer(withoutNamespaceIndex = true)) +public class MyReconciler implements Reconciler { } +``` + +The same option is available on `InformerEventSourceConfiguration.Builder` for event sources, and on +`InformerConfiguration.Builder`. + +This matters most for informers that cache a large number of resources, and in particular together +with a custom [item store](#bounded-caches-for-informers) that keeps only a reduced form of each +resource: the index is keyed independently of what the store does with the resource itself, so +shrinking what is cached does not shrink the index. + +Note that this takes part in the informer pool identity described below: an event source that +removes the index does not share an informer with one that keeps it. If two event sources watch the +same resource type and only one of them sets the option, they end up with two informers, and two +caches of that resource type, which can cost far more memory than the index ever did. + ### Sharing Informers Between Controllers (Informer Pool) {{% alert title="Experimental" color="warning" %}} @@ -374,7 +398,10 @@ Two event sources share an informer when their effective informer configuration - the resource type (or the group/version/kind for generic resources), - the watched namespace, - the label, field and shard selectors, -- the configured [item store](#bounded-caches-for-informers). +- the configured [item store](#bounded-caches-for-informers), +- whether the [namespace index is removed](#removing-the-namespace-index): it cannot be present for + one event source and absent for another on one shared informer, so event sources that disagree on + it are backed by separate informers. The `informerListLimit` is intentionally *not* part of this identity: if two otherwise-equivalent event sources request a different list limit, the existing informer is reused (a warning is logged diff --git a/docs/content/en/docs/documentation/operations/configuration.md b/docs/content/en/docs/documentation/operations/configuration.md index cdfb1b7fdb..20b3ca2003 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"))); ``` +### 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 are officially supported on **Java 25 or later**. They exist from Java 21 on and + the flag does enable them there, but before Java 25 a virtual thread pins its carrier thread + while it is inside a `synchronized` block, which can starve the carrier pool. [JEP + 491](https://openjdk.org/jeps/491), delivered in Java 25, removed that pinning, so this is the + baseline the framework supports. On a JVM without virtual threads at all (below Java 21) 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. +- 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: @@ -265,6 +306,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 (officially supported on Java 25+ 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 | @@ -356,6 +398,7 @@ All controller-level keys are prefixed with `josdk.controller.. | `josdk.controller..informer.label-selector` | `String` | Label selector for the primary resource informer (alias for `label-selector`) | | `josdk.controller..informer.shard-selector` | `String` | Shard selector for the primary resource informer (alias for `shard-selector`) | | `josdk.controller..informer.list-limit` | `Long` | Page size for paginated informer list requests; omit for no pagination | +| `josdk.controller..informer.without-namespace-index` | `Boolean` | Removes the namespace index the informer maintains; defaults to `false`. See [Removing the Namespace Index](../eventing#removing-the-namespace-index) | #### Retry diff --git a/micrometer-support/pom.xml b/micrometer-support/pom.xml index be864bff8e..ae3c4d0be1 100644 --- a/micrometer-support/pom.xml +++ b/micrometer-support/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.6.1-SNAPSHOT + 999-SNAPSHOT micrometer-support diff --git a/migration/pom.xml b/migration/pom.xml index bdbc8fbc4f..cf5143c925 100644 --- a/migration/pom.xml +++ b/migration/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.6.1-SNAPSHOT + 999-SNAPSHOT migration diff --git a/operator-framework-bom/pom.xml b/operator-framework-bom/pom.xml index 5adbefd8d8..0f974400b1 100644 --- a/operator-framework-bom/pom.xml +++ b/operator-framework-bom/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk operator-framework-bom - 5.6.1-SNAPSHOT + 999-SNAPSHOT pom Operator SDK - Bill of Materials Java SDK for implementing Kubernetes operators diff --git a/operator-framework-core/pom.xml b/operator-framework-core/pom.xml index b0ef5cac6f..e0c6031670 100644 --- a/operator-framework-core/pom.xml +++ b/operator-framework-core/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.6.1-SNAPSHOT + 999-SNAPSHOT ../pom.xml @@ -128,7 +128,7 @@ ${git-commit-id-maven-plugin.version} true - ${project.build.outputDirectory}/version.properties + ${project.build.outputDirectory}/operator-sdk-version.properties ^git.build.time$ ^git.commit.id.(abbrev|full)$ 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 db1b9a5fa5..1d750bdc96 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 @@ -16,10 +16,10 @@ package io.javaoperatorsdk.operator.api.config; import java.time.Duration; +import java.util.List; import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Executors; import java.util.function.Consumer; import org.slf4j.Logger; @@ -41,6 +41,7 @@ import io.javaoperatorsdk.operator.api.reconciler.Experimental; import io.javaoperatorsdk.operator.api.reconciler.Reconciler; import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory; +import io.javaoperatorsdk.operator.api.reconciler.dependent.DesiredStateAspect; import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent; import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource; import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig; @@ -153,6 +154,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. @@ -161,13 +168,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(); } /** @@ -201,6 +212,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() { @@ -211,6 +227,12 @@ 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() { @@ -226,6 +248,42 @@ 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. + * + *

Officially supported on Java 25 or later. Virtual threads exist as of Java 21 and are used + * there as well, but before Java 25 a virtual thread pins its carrier thread while inside a + * {@code synchronized} block, which JEP 491 removed in + * Java 25. On a JVM without virtual threads at all, 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. 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 @@ -234,7 +292,8 @@ default Metrics getMetrics() { * processing */ default ExecutorService getExecutorService() { - return Executors.newFixedThreadPool(concurrentReconciliationThreads()); + return ExecutorServiceManager.newBoundedExecutorService( + concurrentReconciliationThreads(), useVirtualThreads()); } /** @@ -244,7 +303,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()); } /** @@ -539,4 +599,18 @@ default InformerPool informerPool() { pool.setConfigurationService(this); return pool; } + + /** + * Retrieves the {@link DesiredStateAspect}s applied to the desired state of all the Kubernetes + * dependent resources managed by the operator. Aspects are applied in the order in which they are + * returned, right after the desired state has been computed, and are typically used to add common + * metadata (such as a label identifying the operator managing the resource) to all the resources + * the operator creates or updates. + * + * @return the list of aspects to apply to computed desired states, empty by default + * @since 5.6.0 + */ + default List desiredStateAspects() { + return List.of(); + } } 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 18a2e3fc38..0f4f74bf37 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 @@ -16,6 +16,8 @@ package io.javaoperatorsdk.operator.api.config; import java.time.Duration; +import java.util.ArrayList; +import java.util.List; import java.util.Optional; import java.util.Set; import java.util.concurrent.ExecutorService; @@ -31,6 +33,7 @@ import io.javaoperatorsdk.operator.api.monitoring.Metrics; import io.javaoperatorsdk.operator.api.reconciler.Experimental; import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory; +import io.javaoperatorsdk.operator.api.reconciler.dependent.DesiredStateAspect; import io.javaoperatorsdk.operator.processing.event.source.informer.pool.InformerPool; @SuppressWarnings({"unused", "UnusedReturnValue"}) @@ -47,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; @@ -59,6 +63,7 @@ public class ConfigurationServiceOverrider { private Boolean useSSAToPatchPrimaryResource; private Boolean cloneSecondaryResourcesWhenGettingFromCache; private InformerPool informerPool; + private List desiredStateAspects; @SuppressWarnings("rawtypes") private DependentResourceFactory dependentResourceFactory; @@ -115,6 +120,19 @@ public ConfigurationServiceOverrider withWorkflowExecutorService( return this; } + /** + * Makes the framework run the tasks it executes concurrently on virtual threads instead of + * platform threads. Officially 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. @@ -229,6 +247,38 @@ public ConfigurationServiceOverrider withInformerPool(InformerPool informerPool) return this; } + /** + * Replaces the {@link DesiredStateAspect}s applied to the desired state of all the Kubernetes + * dependent resources managed by the operator by the specified ones. + * + * @param desiredStateAspects the aspects to apply, in the order in which they should be applied + * @return this {@link ConfigurationServiceOverrider} for chained customization + * @since 5.6.0 + */ + public ConfigurationServiceOverrider withDesiredStateAspects( + List desiredStateAspects) { + this.desiredStateAspects = new ArrayList<>(desiredStateAspects); + return this; + } + + /** + * Appends the specified {@link DesiredStateAspect}s to the already configured ones, which are the + * ones configured on the overridden {@link ConfigurationService} unless {@link + * #withDesiredStateAspects(List)} was called on this overrider first. + * + * @param desiredStateAspects the aspects to append, in the order in which they should be applied + * @return this {@link ConfigurationServiceOverrider} for chained customization + * @since 5.6.0 + */ + public ConfigurationServiceOverrider addDesiredStateAspects( + DesiredStateAspect... desiredStateAspects) { + if (this.desiredStateAspects == null) { + this.desiredStateAspects = new ArrayList<>(original.desiredStateAspects()); + } + this.desiredStateAspects.addAll(List.of(desiredStateAspects)); + return this; + } + public ConfigurationService build() { return new BaseConfigurationService(original.getVersion(), cloner, client) { @Override @@ -286,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) { @@ -383,6 +438,12 @@ public synchronized InformerPool informerPool() { informerPool.setConfigurationService(this); return informerPool; } + + @Override + public List desiredStateAspects() { + return overriddenValueOrDefault( + desiredStateAspects, ConfigurationService::desiredStateAspects); + } }; } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java index 1c1e03c870..0a32bbdcaf 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/ControllerConfigurationOverrider.java @@ -210,6 +210,18 @@ public ControllerConfigurationOverrider withInformerListLimit(Long informerLi return this; } + /** + * Whether to remove the namespace index the underlying informer maintains by default. Note that + * event sources that disagree on this setting do not share an informer. + * + * @param withoutNamespaceIndex true to remove the namespace index, false (the default) to keep it + * @see io.javaoperatorsdk.operator.api.config.informer.Informer#withoutNamespaceIndex() + */ + public ControllerConfigurationOverrider withoutNamespaceIndex(boolean withoutNamespaceIndex) { + config.withoutNamespaceIndex(withoutNamespaceIndex); + return this; + } + public ControllerConfigurationOverrider replacingNamedDependentResourceConfig( String name, Object dependentResourceConfig) { 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..342e07fb9f 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,10 @@ 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()); + // stays on platform threads even when virtual threads are requested: there is no virtual + // thread backed ScheduledExecutorService in the JDK, see VirtualThreads 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/Utils.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Utils.java index 6ad4928c86..4613d9645d 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Utils.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Utils.java @@ -42,6 +42,8 @@ public class Utils { public static final String GENERIC_PARAMETER_TYPE_ERROR_PREFIX = "Couldn't retrieve generic parameter type from "; + public static final String VERSION_PROPERTIES_FILE_NAME = "operator-sdk-version.properties"; + public static final Version VERSION = loadFromProperties(); /** @@ -52,7 +54,9 @@ public class Utils { */ private static Version loadFromProperties() { final var is = - Thread.currentThread().getContextClassLoader().getResourceAsStream("version.properties"); + Thread.currentThread() + .getContextClassLoader() + .getResourceAsStream(VERSION_PROPERTIES_FILE_NAME); final var properties = new Properties(); if (is != null) { @@ -62,7 +66,9 @@ private static Version loadFromProperties() { log.warn("Couldn't load version information: {}", e.getMessage()); } } else { - log.warn("Couldn't find version.properties file. Default version information will be used."); + log.warn( + "Couldn't find {} file. Default version information will be used.", + VERSION_PROPERTIES_FILE_NAME); } Date builtTime; 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..7c7b8edc2b --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/VirtualThreads.java @@ -0,0 +1,209 @@ +/* + * 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.Executor; +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.fabric8.kubernetes.client.KubernetesClientBuilder.ExecutorSupplier; +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. + * + *

There is intentionally no scheduled variant here: the JDK provides no virtual thread backed + * {@link java.util.concurrent.ScheduledExecutorService}, scheduling was deliberately left out of + * virtual threads (the Loom runtime itself uses a platform thread scheduler to unpark timed out + * virtual threads). Running scheduled tasks on virtual threads would mean keeping a platform thread + * scheduler purely for the timing and handing each fired task off to a virtual thread executor, + * which the SDK doesn't do, so {@link ExecutorServiceManager#scheduledExecutorService()} is always + * backed by platform threads. + */ +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, Java 25 or later is officially supported). 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" + + " (Java 25 or later is officially supported)"); + } + 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 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; + 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/main/java/io/javaoperatorsdk/operator/api/config/informer/Informer.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Informer.java index 04f97902d3..69999ef04f 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Informer.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/Informer.java @@ -31,6 +31,7 @@ import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_COMPARABLE_RESOURCE_VERSION; import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_FOLLOW_CONTROLLER_NAMESPACE_CHANGES; import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_GHOST_RESOURCE_CHECK_INTERVAL_MILLIS; +import static io.javaoperatorsdk.operator.api.reconciler.Constants.DEFAULT_WITHOUT_NAMESPACE_INDEX; import static io.javaoperatorsdk.operator.api.reconciler.Constants.NO_LONG_VALUE_SET; import static io.javaoperatorsdk.operator.api.reconciler.Constants.NO_VALUE_SET; @@ -153,6 +154,23 @@ */ boolean comparableResourceVersions() default DEFAULT_COMPARABLE_RESOURCE_VERSION; + /** + * Whether to remove the namespace index that the underlying informer maintains by default. + * + *

The framework never reads that index, it only reads the indexes registered through {@link + * io.javaoperatorsdk.operator.processing.event.source.IndexerResourceCache#addIndexers}, so + * dropping it saves an entry per cached resource. It is worth setting when the informer caches a + * large number of resources and nothing looks them up by namespace, in particular together with a + * custom {@link #itemStore()} that keeps only a reduced form of each resource. + * + *

Note that this makes the informer a distinct one for pooling purposes: event sources that + * disagree on this setting do not share an informer, because the index cannot be present for one + * of them and absent for the other. + * + * @since 5.7.0 + */ + boolean withoutNamespaceIndex() default DEFAULT_WITHOUT_NAMESPACE_INDEX; + /** * @deprecated Ghost resource checking is now triggered by the informer's onList callback. This * setting is no longer used. diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java index 9fe25c999d..106f5898c5 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerConfiguration.java @@ -58,6 +58,7 @@ public class InformerConfiguration { private Long informerListLimit; private FieldSelector fieldSelector; private Boolean comparableResourceVersions; + private Boolean withoutNamespaceIndex; protected InformerConfiguration( Class resourceClass, @@ -75,6 +76,7 @@ protected InformerConfiguration( Long informerListLimit, FieldSelector fieldSelector, Boolean comparableResourceVersions, + Boolean withoutNamespaceIndex, // TODO for removal in major release Duration ghostResourceCacheCheckInterval) { this(resourceClass, resourceGroupVersionKind); @@ -91,6 +93,7 @@ protected InformerConfiguration( this.informerListLimit = informerListLimit; this.fieldSelector = fieldSelector; this.comparableResourceVersions = comparableResourceVersions; + this.withoutNamespaceIndex = withoutNamespaceIndex; } private InformerConfiguration(Class resourceClass, GroupVersionKind resourceGroupVersionKind) { @@ -140,6 +143,7 @@ public static InformerConfiguration.Builder builder( original.informerListLimit, original.fieldSelector, original.comparableResourceVersions, + original.withoutNamespaceIndex, null) .builder; } @@ -332,6 +336,15 @@ public boolean isComparableResourceVersions() { return comparableResourceVersions; } + /** + * Whether the namespace index the underlying informer maintains by default is removed. + * + * @see Informer#withoutNamespaceIndex() + */ + public boolean isWithoutNamespaceIndex() { + return withoutNamespaceIndex; + } + @SuppressWarnings("UnusedReturnValue") public class Builder { @@ -349,6 +362,9 @@ public InformerConfiguration buildForController() { if (comparableResourceVersions == null) { comparableResourceVersions = DEFAULT_COMPARABLE_RESOURCE_VERSION; } + if (withoutNamespaceIndex == null) { + withoutNamespaceIndex = DEFAULT_WITHOUT_NAMESPACE_INDEX; + } return InformerConfiguration.this; } @@ -364,6 +380,9 @@ public InformerConfiguration build() { if (comparableResourceVersions == null) { comparableResourceVersions = DEFAULT_COMPARABLE_RESOURCE_VERSION; } + if (withoutNamespaceIndex == null) { + withoutNamespaceIndex = DEFAULT_WITHOUT_NAMESPACE_INDEX; + } return InformerConfiguration.this; } @@ -417,6 +436,7 @@ public InformerConfiguration.Builder initFromAnnotation( .map(f -> new FieldSelector.Field(f.path(), f.value(), f.negated())) .toList())); withComparableResourceVersions(informerConfig.comparableResourceVersions()); + withoutNamespaceIndex(informerConfig.withoutNamespaceIndex()); } return this; } @@ -533,6 +553,16 @@ private static boolean isEmpty(FieldSelector fieldSelector) { || fieldSelector.getFields().isEmpty(); } + /** + * Whether to remove the namespace index the underlying informer maintains by default. + * + * @see Informer#withoutNamespaceIndex() + */ + public Builder withoutNamespaceIndex(boolean withoutNamespaceIndex) { + InformerConfiguration.this.withoutNamespaceIndex = withoutNamespaceIndex; + return this; + } + public Builder withComparableResourceVersions(boolean comparableResourceVersions) { InformerConfiguration.this.comparableResourceVersions = comparableResourceVersions; return this; diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java index 9bd6f84d06..d40469de4c 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/informer/InformerEventSourceConfiguration.java @@ -292,6 +292,11 @@ public Builder withComparableResourceVersion(boolean comparableResourceVersio return this; } + public Builder withoutNamespaceIndex(boolean withoutNamespaceIndex) { + config.withoutNamespaceIndex(withoutNamespaceIndex); + return this; + } + @Deprecated(forRemoval = true) public Builder withGhostResourceCacheCheckInterval( Duration ghostResourceCacheCheckInterval) { @@ -316,6 +321,8 @@ public void updateFrom(InformerConfiguration informerConfig) { .withOnDeleteFilter(informerConfig.getOnDeleteFilter()) .withGenericFilter(informerConfig.getGenericFilter()) .withInformerListLimit(informerConfig.getInformerListLimit()) + .withComparableResourceVersions(informerConfig.isComparableResourceVersions()) + .withoutNamespaceIndex(informerConfig.isWithoutNamespaceIndex()) .withFieldSelector(informerConfig.getFieldSelector()); } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorder.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorder.java index 2e7023623d..67764b7146 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorder.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorder.java @@ -24,6 +24,7 @@ import java.time.temporal.ChronoUnit; import java.util.HexFormat; import java.util.Objects; +import java.util.regex.Pattern; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -48,10 +49,11 @@ * can be overridden, see {@link * io.javaoperatorsdk.operator.api.config.ConfigurationService#clusterScopedEventNamespace()}. * - *

Events are named deterministically, after the object they are about plus a hash of everything - * that identifies the event, so that recording the same event again resolves to the event already - * recorded for it. Repeat occurrences are then counted on that event rather than recorded as copies - * of it, see {@link DefaultEventSink}. + *

By default, events are named deterministically, after the object they are about plus a hash of + * everything that identifies the event, so that recording the same event again resolves to the + * event already recorded for it. Repeat occurrences are then counted on that event rather than + * recorded as copies of it, see {@link DefaultEventSink}. How events aggregate, how they are named + * and whether they carry an owner reference can be configured, see {@link #builder(EventSink)}. */ public class DefaultEventRecorder implements EventRecorder { @@ -73,10 +75,79 @@ public class DefaultEventRecorder implements EventRecorder { private static final int IDENTITY_HASH_LENGTH = 32; + /** What the API server accepts as an object name, see RFC 1123 on DNS subdomains. */ + private static final Pattern RFC_1123_SUBDOMAIN = + Pattern.compile("[a-z0-9]([-a-z0-9]*[a-z0-9])?(\\.[a-z0-9]([-a-z0-9]*[a-z0-9])?)*"); + private final EventSink sink; + private final EventNamingStrategy namingStrategy; + private final EventKeyStrategy keyStrategy; + private final boolean ownerReference; public DefaultEventRecorder(EventSink sink) { + this(sink, EventNamingStrategy.none(), EventKeyStrategy.none(), false); + } + + private DefaultEventRecorder( + EventSink sink, + EventNamingStrategy namingStrategy, + EventKeyStrategy keyStrategy, + boolean ownerReference) { this.sink = sink; + this.namingStrategy = namingStrategy; + this.keyStrategy = keyStrategy; + this.ownerReference = ownerReference; + } + + public static Builder builder(EventSink sink) { + return new Builder(sink); + } + + /** Builder for {@link DefaultEventRecorder}. */ + public static final class Builder { + + private final EventSink sink; + private EventNamingStrategy namingStrategy = EventNamingStrategy.none(); + private EventKeyStrategy keyStrategy = EventKeyStrategy.none(); + private boolean ownerReference = false; + + private Builder(EventSink sink) { + this.sink = Objects.requireNonNull(sink, "sink must not be null"); + } + + /** The strategy naming recorded events, see {@link EventNamingStrategy}. */ + public Builder namingStrategy(EventNamingStrategy namingStrategy) { + this.namingStrategy = + Objects.requireNonNull(namingStrategy, "namingStrategy must not be null"); + return this; + } + + /** + * The strategy deriving the default aggregation key of records that do not set one, see {@link + * EventKeyStrategy}. The key is ignored for events whose name is set by the record or resolved + * by the naming strategy, see {@link EventNamingStrategy}. + */ + public Builder keyStrategy(EventKeyStrategy keyStrategy) { + this.keyStrategy = Objects.requireNonNull(keyStrategy, "keyStrategy must not be null"); + return this; + } + + /** + * When set, recorded events carry an {@code ownerReference} to the object they are about. The + * reference expresses ownership for tooling that reads it; note that the Kubernetes garbage + * collector ignores events, so it does not cause cascade deletion, events expire through the + * event TTL either way. Records can override this per event via {@link + * EventRecord.Builder#ownedByRegarding(boolean)}. The reference is only set when the object + * already has a uid. + */ + public Builder ownerReference(boolean ownerReference) { + this.ownerReference = ownerReference; + return this; + } + + public DefaultEventRecorder build() { + return new DefaultEventRecorder(sink, namingStrategy, keyStrategy, ownerReference); + } } /** @@ -111,14 +182,17 @@ private static String resolve() { public void record(EventRecord event, Context context) { Objects.requireNonNull(context, "the context of the reconciliation must not be null"); Objects.requireNonNull(event, "event must not be null"); + Event assembled = null; try { - sink.emit(toEvent(context, event), context); + assembled = toEvent(context, event); + sink.emit(assembled, context); } catch (Exception e) { // recording an event must never break the caller: a controller that fails to reconcile // because it could not write an event is strictly worse than one that records nothing log.warn( - "Could not record {} event with reason {} for resource {} in namespace {}", + "Could not record {} event named {} with reason {} for resource {} in namespace {}", event.type(), + assembled != null ? assembled.getMetadata().getName() : "unknown", event.reason(), context.getPrimaryResource().getMetadata().getName(), context.getPrimaryResource().getMetadata().getNamespace(), @@ -164,6 +238,23 @@ protected Event toEvent(Context context, EventRecord record) { .withNewSource() .withComponent(record.reportingComponent().orElse(controllerName)) .endSource(); + boolean ownedByRegarding = record.ownedByRegarding().orElse(ownerReference); + if (ownedByRegarding && regarding.getMetadata().getUid() == null) { + log.debug( + "Not setting the owner reference on the event about {}: the object has no uid yet", + regarding.getMetadata().getName()); + } + if (ownedByRegarding && regarding.getMetadata().getUid() != null) { + builder + .editMetadata() + .addNewOwnerReference() + .withApiVersion(regarding.getApiVersion()) + .withKind(regarding.getKind()) + .withName(regarding.getMetadata().getName()) + .withUid(regarding.getMetadata().getUid()) + .endOwnerReference() + .endMetadata(); + } record.action().ifPresent(builder::withAction); return builder.build(); } @@ -181,17 +272,56 @@ private String eventNamespace(HasMetadata regarding, Context context) { CLUSTER_SCOPED_EVENT_NAMESPACE); } + private String eventName(HasMetadata regarding, EventRecord record, String reportingController) { + return record + .name() + .filter(name -> !name.isBlank()) + .or(() -> namingStrategy.nameFor(regarding, record).filter(name -> !name.isBlank())) + .map(DefaultEventRecorder::truncateToMaxNameLength) + .filter(DefaultEventRecorder::isValidEventName) + .orElseGet(() -> identityHashName(regarding, record, reportingController)); + } + + private static boolean isValidEventName(String name) { + if (RFC_1123_SUBDOMAIN.matcher(name).matches()) { + return true; + } + log.warn( + "Falling back to the default event name: {} is not a valid RFC 1123 DNS subdomain", name); + return false; + } + + private static String truncateToMaxNameLength(String name) { + if (name.length() <= MAX_NAME_LENGTH) { + return name; + } + var truncated = name.substring(0, MAX_NAME_LENGTH); + while (truncated.endsWith("-") || truncated.endsWith(".")) { + truncated = truncated.substring(0, truncated.length() - 1); + } + log.warn( + "Truncated the name of event {} to {} to stay within the Kubernetes name limit", + name, + truncated); + return truncated; + } + /** * Names events {@code .}, following the convention of the Go client, hashing * everything that makes two events the same event: the object, the type, the reason, the - * reporting component and, unless the record sets a {@link EventRecord#key()}, the message. The - * name is therefore stable across occurrences, which is what lets the sink recognise a repeat, - * and stays so across operator restarts and between replicas, unlike a name remembered in memory. + * reporting component and, unless the record sets a {@link EventRecord#key()} or the recorder is + * built with a default {@link EventKeyStrategy}, the message. The name is therefore stable across + * occurrences, which is what lets the sink recognise a repeat, and stays so across operator + * restarts and between replicas, unlike a name remembered in memory. * *

The object is identified by its uid, with the kind as a fallback for objects that do not * have one yet, such as a dependent resource that has only been built so far. + * + *

This is the fallback when the record does not set a name and the naming strategy resolves to + * nothing, see {@link EventNamingStrategy}. */ - private String eventName(HasMetadata regarding, EventRecord record, String reportingController) { + private String identityHashName( + HasMetadata regarding, EventRecord record, String reportingController) { var metadata = regarding.getMetadata(); var identity = String.join( @@ -201,7 +331,10 @@ private String eventName(HasMetadata regarding, EventRecord record, String repor record.type().value(), record.reason(), record.reportingComponent().orElse(reportingController), - record.key().orElseGet(() -> requireNonNullElse(record.message(), ""))); + record + .key() + .or(() -> keyStrategy.keyFor(regarding, record)) + .orElseGet(() -> requireNonNullElse(record.message(), ""))); var suffix = "." + identityDigest(identity); var prefix = metadata.getName(); diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventKeyStrategy.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventKeyStrategy.java new file mode 100644 index 0000000000..632ff6ea6a --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventKeyStrategy.java @@ -0,0 +1,55 @@ +/* + * 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.event; + +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.javaoperatorsdk.operator.api.reconciler.Experimental; + +import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE; + +/** + * Derives the default aggregation key of an event, used when the {@link EventRecord} does not set + * one explicitly. The key identifies an event among the events about the same object, so that + * repeated occurrences resolve to the same event rather than to one event each, see {@link + * EventRecord#key()}. + * + *

An empty result leaves the record without a default key, which keeps the message part of the + * event identity. + * + *

Implementations are called from concurrent reconciliations and must be thread safe. + */ +@Experimental(API_MIGHT_CHANGE) +@FunctionalInterface +public interface EventKeyStrategy { + + Optional keyFor(HasMetadata regarding, EventRecord record); + + /** No default key: the message stays part of the event identity. */ + static EventKeyStrategy none() { + return (regarding, record) -> Optional.empty(); + } + + /** + * Aggregates by event type and reason: all occurrences of a reason resolve to one event whose + * count grows and whose message is replaced with the latest one. The right choice for events that + * report a current state rather than individual occurrences. + */ + static EventKeyStrategy byReason() { + return (regarding, record) -> Optional.of(record.type().value() + "/" + record.reason()); + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventNamingStrategy.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventNamingStrategy.java new file mode 100644 index 0000000000..80f6f33e82 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventNamingStrategy.java @@ -0,0 +1,51 @@ +/* + * 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.event; + +import java.util.Optional; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.javaoperatorsdk.operator.api.reconciler.Experimental; + +import static io.javaoperatorsdk.operator.api.reconciler.Experimental.API_MIGHT_CHANGE; + +/** + * Names the event recorded about an object. The name is what the sink looks a recorded event up by, + * so it is also the aggregation identity: two records resolving to the same name are counted as + * occurrences of one event. A name must therefore be unique among the events it should not + * aggregate with, and stable across operator restarts and replicas. + * + *

A name must be a valid RFC 1123 DNS subdomain: at most 253 lowercase alphanumeric characters, + * {@code -} or {@code .}, starting and ending with an alphanumeric character. Names longer than the + * limit are truncated. A name derived from the object and a fixed lowercase suffix (such as {@code + * -status-report}) satisfies all of this by construction. + * + *

An empty result or an invalid name falls back to the default {@code .} + * name, rather than the event being lost to the API server rejecting the name. + * + *

Implementations are called from concurrent reconciliations and must be thread safe. + */ +@Experimental(API_MIGHT_CHANGE) +@FunctionalInterface +public interface EventNamingStrategy { + + Optional nameFor(HasMetadata regarding, EventRecord record); + + /** No custom naming: every event gets the default {@code .} name. */ + static EventNamingStrategy none() { + return (regarding, record) -> Optional.empty(); + } +} diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventRecord.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventRecord.java index e7b736abc0..6d98b7f1fe 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventRecord.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/event/EventRecord.java @@ -32,7 +32,9 @@ public final class EventRecord { private final EventType type; private final String reason; private final String message; + private final String name; private final String key; + private final Boolean ownedByRegarding; private final String action; private final String reportingComponent; private final Map labels; @@ -42,11 +44,13 @@ private EventRecord(Builder builder) { this.type = builder.type; this.reason = builder.reason; this.message = builder.message; + this.name = builder.name; this.key = builder.key; this.action = builder.action; this.reportingComponent = builder.reportingComponent; this.labels = Map.copyOf(builder.labels); this.annotations = Map.copyOf(builder.annotations); + this.ownedByRegarding = builder.ownedByRegarding; } public static Builder builder() { @@ -75,14 +79,34 @@ public String message() { return message; } + /** + * The name of the recorded event, overriding the recorder's naming. The name is the aggregation + * identity and must be a valid RFC 1123 DNS subdomain, see {@link EventNamingStrategy}; an + * invalid name falls back to the default name. A blank name is treated as unset. + */ + public Optional name() { + return Optional.ofNullable(name); + } + /** * Identifies this event among the events about the same object, so that repeated occurrences * resolve to the same event rather than to one event each. + * + *

The key is ignored when the record sets a {@link #name()} or the recorder's naming strategy + * resolves one: the name is then the aggregation identity on its own. */ public Optional key() { return Optional.ofNullable(key); } + /** + * Whether the recorded event carries an {@code ownerReference} to the object it is about. When + * empty, the recorder's own setting applies. + */ + public Optional ownedByRegarding() { + return Optional.ofNullable(ownedByRegarding); + } + /** * The action taken or failed regarding the involved object, if any. Optional, and only meaningful * for consumers that read the {@code action} field of the event. @@ -119,7 +143,9 @@ public static final class Builder { private EventType type = EventType.NORMAL; private String reason; private String message; + private String name; private String key; + private Boolean ownedByRegarding; private String action; private String reportingComponent; private final Map labels = new HashMap<>(); @@ -142,12 +168,27 @@ public Builder message(String message) { return this; } + /** Sets the name of the recorded event, see {@link EventRecord#name()}. */ + public Builder name(String name) { + this.name = name; + return this; + } + /** Sets the key identifying this event, see {@link EventRecord#key()}. */ public Builder key(String key) { this.key = key; return this; } + /** + * Sets whether this event is owned by the object it is about, see {@link + * EventRecord#ownedByRegarding()}. + */ + public Builder ownedByRegarding(boolean ownedByRegarding) { + this.ownedByRegarding = ownedByRegarding; + return this; + } + public Builder action(String action) { this.action = action; return this; diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Constants.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Constants.java index fad19d2021..2ae2de8b4f 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Constants.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/Constants.java @@ -43,6 +43,7 @@ public final class Constants { public static final String CONTROLLER_NAME = "controller.name"; public static final boolean DEFAULT_FOLLOW_CONTROLLER_NAMESPACE_CHANGES = true; public static final boolean DEFAULT_COMPARABLE_RESOURCE_VERSION = true; + public static final boolean DEFAULT_WITHOUT_NAMESPACE_INDEX = false; @Deprecated(forRemoval = true) public static final long DEFAULT_GHOST_RESOURCE_CHECK_INTERVAL_MILLIS = 3 * 60 * 1000; 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 e399f7fdfd..50e61526c4 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 @@ -31,6 +31,7 @@ import io.javaoperatorsdk.operator.api.config.ControllerConfiguration; import io.javaoperatorsdk.operator.api.event.ResourceEventRecorder; import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResource; +import io.javaoperatorsdk.operator.api.reconciler.dependent.DesiredStateAspect; import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.DefaultManagedWorkflowAndDependentResourceContext; import io.javaoperatorsdk.operator.api.reconciler.dependent.managed.ManagedWorkflowAndDependentResourceContext; import io.javaoperatorsdk.operator.processing.Controller; @@ -258,6 +259,25 @@ public R getOrComputeDesiredStateFor( DependentResource dependentResource, Function desiredStateComputer) { return (R) desiredStates.computeIfAbsent( - dependentResource, ignored -> desiredStateComputer.apply(getPrimaryResource())); + dependentResource, + ignored -> { + final var desired = desiredStateComputer.apply(getPrimaryResource()); + applyDesiredStateAspects(desired, dependentResource); + return desired; + }); + } + + /** + * Applies the globally configured {@link DesiredStateAspect}s, in configuration order, to the + * freshly computed desired state. Aspects only apply to Kubernetes resources, external dependent + * resources are therefore left untouched. + */ + private void applyDesiredStateAspects(Object desired, DependentResource dependentResource) { + if (desired instanceof HasMetadata hasMetadata) { + controllerConfiguration + .getConfigurationService() + .desiredStateAspects() + .forEach(aspect -> aspect.apply(hasMetadata, dependentResource, this)); + } } } diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DesiredStateAspect.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DesiredStateAspect.java new file mode 100644 index 0000000000..826a15cb66 --- /dev/null +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/reconciler/dependent/DesiredStateAspect.java @@ -0,0 +1,55 @@ +/* + * 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.reconciler.dependent; + +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.javaoperatorsdk.operator.api.config.ConfigurationService; +import io.javaoperatorsdk.operator.api.reconciler.Context; + +/** + * A cross-cutting hook applied to the desired state of every Kubernetes {@link DependentResource} + * managed by the operator, typically used to add common metadata (labels or annotations) marking + * the resources the operator manages. + * + *

Aspects are registered globally on the {@link ConfigurationService} and are applied, in + * registration order, right after the desired state has been computed and before it is matched + * against, created or updated. This means modifications performed by an aspect are taken into + * account when determining whether the actual resource matches its desired state, so that changing + * an aspect triggers an update of the associated secondary resources. + * + *

The desired state is computed at most once per reconciliation and cached in the {@link + * Context}, so aspects are also called at most once per dependent resource and reconciliation. + * Aspects are only applied to dependent resources whose desired state is a {@link HasMetadata}, + * i.e. they are not called for external (non-Kubernetes) dependent resources. + * + *

Implementations are expected to mutate the provided desired state in place and must be + * thread-safe as they can be called concurrently for different primary resources. + * + * @see ConfigurationService#desiredStateAspects() + */ +@FunctionalInterface +public interface DesiredStateAspect { + + /** + * Applies this aspect to the specified, freshly computed desired state. + * + * @param desired the desired state to modify in place + * @param dependentResource the {@link DependentResource} the desired state was computed for + * @param context the {@link Context} of the current reconciliation, from which the primary + * resource can be retrieved using {@link Context#getPrimaryResource()} + */ + void apply(HasMetadata desired, DependentResource dependentResource, Context context); +} 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 827961b77f..2428eeba96 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 @@ -117,7 +117,10 @@ public R update(R actual, R desired, P primary, Context

context) { @Override public Result match(R resource, P primary, Context

context) { - return bulkDependentResource.match(resource, desired, primary, context); + // retrieve the desired state via the context so that it is processed the same way as for + // non-bulk dependents, in particular so that configured DesiredStateAspects are applied + // before matching + return bulkDependentResource.match(resource, getOrComputeDesired(context), primary, context); } @Override diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java index 6caf39ccd9..ae6694cfe4 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManager.java @@ -169,7 +169,8 @@ private InformerClassifier getClassifier(String namespaceIdentifier) { configuration.getInformerConfig().getResourceGroupVersionKind(), configuration.getInformerConfig().getFieldSelector(), configuration.getInformerConfig().getInformerListLimit(), - configuration.getInformerConfig().getItemStore()); + configuration.getInformerConfig().getItemStore(), + configuration.getInformerConfig().isWithoutNamespaceIndex()); } private KubernetesClient getTargetClient() { diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java index 4dc1920955..21464b378b 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPool.java @@ -71,7 +71,7 @@ public void setConfigurationService(ConfigurationService configurationService) { */ public abstract long numberOfInformersForResource(Class resourceClass); - @SuppressWarnings({"rawtypes", "unchecked"}) + @SuppressWarnings({"rawtypes", "unchecked", "resource"}) protected SharedIndexInformer createInformer(InformerClassifier classifier) { var client = classifier.client(); @@ -116,6 +116,11 @@ protected SharedIndexInformer createInformer(InformerClassifier classifier) { Optional.ofNullable(classifier.itemStore()).ifPresent(informer::itemStore); + if (classifier.withoutNamespaceIndex()) { + // the framework only reads the indexes registered through the resource cache, never this one + informer.removeNamespaceIndex(); + } + configurationService .getInformerStoppedHandler() .ifPresent( diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java index b85b7d9f0b..24dcdba4c8 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPool.java @@ -43,6 +43,14 @@ public SharedIndexInformer getInformer( synchronized (this) { var pooled = informers.get(classifier); if (pooled == null) { + if (informers.keySet().stream() + .anyMatch(existing -> existing.differsOnlyByNamespaceIndex(classifier))) { + log.warn( + "Creating a second informer for classifier {} that differs from an existing one only" + + " by withoutNamespaceIndex, so the resource type is cached twice. Set the" + + " option the same way on both to share one informer.", + classifier); + } informer = createInformer(classifier); informers.put(classifier, new PooledInformer(informer, new AtomicInteger(1))); log.debug( diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifier.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifier.java index e4023a93e9..5f4cfae87f 100644 --- a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifier.java +++ b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifier.java @@ -36,6 +36,9 @@ * limit still share an informer; the limit of whichever classifier created the informer is * kept (a pool is expected to warn about this, see {@link * #differsOnlyByInformerListLimit(InformerClassifier)}). + *

  • {@link #withoutNamespaceIndex()} on the other hand is part of the + * identity: the namespace index cannot be present for one event source and absent for another + * that shares the same informer, so event sources disagreeing on it get separate informers. *
  • Indexers are not part of the classifier at all: they are registered on the informer under a * name qualified with the event source that added them, so those of different event sources * can live side by side on a shared informer without colliding. @@ -61,7 +64,8 @@ public record InformerClassifier( GroupVersionKind groupVersionKind, FieldSelector fieldSelector, Long informerListLimit, - ItemStore itemStore) { + ItemStore itemStore, + boolean withoutNamespaceIndex) { @Override public boolean equals(Object o) { @@ -71,6 +75,16 @@ public boolean equals(Object o) { if (!(o instanceof InformerClassifier that)) { return false; } + return equalsIgnoringNamespaceIndex(that) + && withoutNamespaceIndex == that.withoutNamespaceIndex; + } + + /** + * Equality of everything the identity is made of except {@link #withoutNamespaceIndex()}, so that + * {@link #equals(Object)} and {@link #differsOnlyByNamespaceIndex(InformerClassifier)} cannot + * drift apart when a component is added. + */ + private boolean equalsIgnoringNamespaceIndex(InformerClassifier that) { return client == that.client && Objects.equals(labelSelector, that.labelSelector) && Objects.equals(shardSelector, that.shardSelector) @@ -91,7 +105,8 @@ public int hashCode() { resourceClass, groupVersionKind, fieldSelector, - itemStore); + itemStore, + withoutNamespaceIndex); } /** @@ -123,6 +138,8 @@ public String toString() { + informerListLimit + ", itemStore=" + itemStore + + ", withoutNamespaceIndex=" + + withoutNamespaceIndex + "]"; } @@ -140,4 +157,14 @@ private String masterUrl() { public boolean differsOnlyByInformerListLimit(InformerClassifier other) { return equals(other) && !Objects.equals(informerListLimit, other.informerListLimit); } + + /** + * Checks whether this classifier and the other are equal in every attribute except for {@link + * #withoutNamespaceIndex()}, which differs between them. Unlike the list limit, that setting is + * part of the identity, so such a pair is served by two informers rather than one. + */ + public boolean differsOnlyByNamespaceIndex(InformerClassifier other) { + return equalsIgnoringNamespaceIndex(other) + && withoutNamespaceIndex != other.withoutNamespaceIndex; + } } 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 4356395618..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 @@ -16,6 +16,7 @@ package io.javaoperatorsdk.operator.api.config; import java.time.Duration; +import java.util.List; import java.util.Optional; import java.util.Set; import java.util.concurrent.Executors; @@ -33,6 +34,7 @@ import io.javaoperatorsdk.operator.api.monitoring.Metrics; import io.javaoperatorsdk.operator.api.reconciler.Context; import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory; +import io.javaoperatorsdk.operator.api.reconciler.dependent.DesiredStateAspect; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.jupiter.api.Assertions.assertNotEquals; @@ -150,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() { @@ -187,4 +196,36 @@ void clusterScopedEventNamespaceDefaultsToTheDefaultNamespaceAndCanBeOverridden( .clusterScopedEventNamespace()) .isEqualTo("operator-ns"); } + + @Test + void desiredStateAspectsAreEmptyByDefaultAndCanBeOverridden() { + assertThat(config.desiredStateAspects()).isEmpty(); + + final DesiredStateAspect first = (desired, dependentResource, context) -> {}; + final DesiredStateAspect second = (desired, dependentResource, context) -> {}; + + assertThat( + new ConfigurationServiceOverrider(config) + .withDesiredStateAspects(List.of(first, second)) + .build() + .desiredStateAspects()) + .containsExactly(first, second); + } + + @Test + void desiredStateAspectsCanBeAppendedToAlreadyConfiguredOnes() { + final DesiredStateAspect first = (desired, dependentResource, context) -> {}; + final DesiredStateAspect second = (desired, dependentResource, context) -> {}; + final DesiredStateAspect third = (desired, dependentResource, context) -> {}; + + final var configWithAspect = + new ConfigurationServiceOverrider(config).withDesiredStateAspects(List.of(first)).build(); + + assertThat( + new ConfigurationServiceOverrider(configWithAspect) + .addDesiredStateAspects(second, third) + .build() + .desiredStateAspects()) + .containsExactly(first, second, third); + } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerConfigurationTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerConfigurationTest.java index 16e5ab578b..ef1bfafd24 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerConfigurationTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerConfigurationTest.java @@ -75,6 +75,28 @@ void nullLabelSelectorByDefault() { assertNull(informerConfig.getLabelSelector()); } + @Test + void keepsTheNamespaceIndexByDefault() { + final var informerConfig = InformerConfiguration.builder(ConfigMap.class).build(); + assertFalse(informerConfig.isWithoutNamespaceIndex()); + } + + @Test + void keepsTheNamespaceIndexByDefaultForController() { + final var informerConfig = InformerConfiguration.builder(ConfigMap.class).buildForController(); + assertFalse(informerConfig.isWithoutNamespaceIndex()); + } + + @Test + void withoutNamespaceIndexIsCarriedOverWhenCopyingTheConfiguration() { + final var original = + InformerConfiguration.builder(ConfigMap.class).withoutNamespaceIndex(true).build(); + + final var copy = InformerConfiguration.builder(original).build(); + + assertTrue(copy.isWithoutNamespaceIndex()); + } + @Test void nullShardSelectorByDefault() { final var informerConfig = InformerConfiguration.builder(ConfigMap.class).build(); diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerEventSourceConfigurationTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerEventSourceConfigurationTest.java new file mode 100644 index 0000000000..9e43b288cc --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/InformerEventSourceConfigurationTest.java @@ -0,0 +1,42 @@ +/* + * 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 org.junit.jupiter.api.Test; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.Secret; +import io.javaoperatorsdk.operator.api.config.informer.InformerConfiguration; +import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; + +import static org.junit.jupiter.api.Assertions.assertFalse; + +class InformerEventSourceConfigurationTest { + + @Test + void updateFromKeepsComparableResourceVersions() { + // the only caller is KubernetesDependentResource#createEventSource, so dropping it here means + // the setting silently has no effect on a dependent resource + var builder = InformerEventSourceConfiguration.from(ConfigMap.class, Secret.class); + + builder.updateFrom( + InformerConfiguration.builder(ConfigMap.class) + .withComparableResourceVersions(false) + .build()); + + assertFalse(builder.build().getInformerConfig().isComparableResourceVersions()); + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VersionTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VersionTest.java index a8202dd0b3..7fff961ff7 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VersionTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VersionTest.java @@ -18,6 +18,8 @@ import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; class VersionTest { @@ -28,4 +30,13 @@ void versionShouldReturnTheSameResultFromMavenAndProperties() { assertEquals(versionFromProperties, versionFromMaven); } + + @Test + void versionShouldBeLoadedFromTheGeneratedPropertiesFile() { + assertNotNull( + Thread.currentThread() + .getContextClassLoader() + .getResource(Utils.VERSION_PROPERTIES_FILE_NAME)); + assertNotEquals(Version.UNKNOWN.getCommit(), Utils.VERSION.getCommit()); + } } 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..c28be4b4fc --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/config/VirtualThreadsTest.java @@ -0,0 +1,197 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.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 + @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(); + } + + @Test + void defaultKubernetesClientKeepsPlatformThreadsWhenVirtualThreadsAreNotRequested() + throws Exception { + try (var client = defaultKubernetesClient(false)) { + assertThat(runsTasksOnVirtualThread(client)).isFalse(); + } + } + + @Test + @EnabledForJreRange(min = JRE.JAVA_21) + 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/api/event/DefaultEventRecorderTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorderTest.java index eb713e5d74..388217b9be 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorderTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/api/event/DefaultEventRecorderTest.java @@ -226,6 +226,14 @@ void reasonIsRequired() { .isThrownBy(() -> EventRecord.builder().message("no reason given").build()); } + @Test + void alwaysDerivesTheSameDefaultNameForTheSameEvent() { + recorder.record(EventRecord.normal("Created", "created"), context(configMap())); + + assertThat(emitted.get(0).getMetadata().getName()) + .isEqualTo("test1.3c699548f37ff9cd6d2a786f64a27228"); + } + @Test void namesEventsWithADnsSafeHashSuffix() { recorder.record(EventRecord.normal("Created", "created"), context(configMap())); @@ -249,6 +257,267 @@ void givesEventsWhoseMessagesCollideUnderStringHashCodeDistinctNames() { .isNotEqualTo(emitted.get(1).getMetadata().getName()); } + @Test + void takesTheMessageOutOfTheEventIdentityWithADefaultKeyStrategy() { + var recorder = + DefaultEventRecorder.builder((event, context) -> emitted.add(event)) + .keyStrategy(EventKeyStrategy.byReason()) + .build(); + var context = context(configMap()); + + recorder.record(EventRecord.warning("Failed", "first message"), context); + recorder.record(EventRecord.warning("Failed", "second message"), context); + + assertThat(emitted.get(0).getMetadata().getName()) + .isEqualTo(emitted.get(1).getMetadata().getName()); + } + + @Test + void prefersThePerRecordKeyOverTheDefaultKeyStrategy() { + var recorder = + DefaultEventRecorder.builder((event, context) -> emitted.add(event)) + .keyStrategy(EventKeyStrategy.byReason()) + .build(); + var context = context(configMap()); + + recorder.record(EventRecord.warning("Failed", "message"), context); + recorder.record( + EventRecord.builder() + .type(EventType.WARNING) + .reason("Failed") + .message("message") + .key("another aggregate") + .build(), + context); + + assertThat(emitted.get(0).getMetadata().getName()) + .isNotEqualTo(emitted.get(1).getMetadata().getName()); + } + + @Test + void keepsTheMessageInTheEventIdentityWithoutADefaultKeyStrategy() { + var context = context(configMap()); + + recorder.record(EventRecord.warning("Failed", "first message"), context); + recorder.record(EventRecord.warning("Failed", "second message"), context); + + assertThat(emitted.get(0).getMetadata().getName()) + .isNotEqualTo(emitted.get(1).getMetadata().getName()); + } + + @Test + void keepsEventsWithTheSameReasonButDifferentTypesApartUnderByReason() { + var recorder = + DefaultEventRecorder.builder((event, context) -> emitted.add(event)) + .keyStrategy(EventKeyStrategy.byReason()) + .build(); + var context = context(configMap()); + + recorder.record(EventRecord.normal("Flipped", "message"), context); + recorder.record(EventRecord.warning("Flipped", "message"), context); + + assertThat(emitted.get(0).getMetadata().getName()) + .isNotEqualTo(emitted.get(1).getMetadata().getName()); + } + + @Test + void setsTheOwnerReferenceToTheInvolvedObjectWhenOwningEventsByRegarding() { + var recorder = + DefaultEventRecorder.builder((event, context) -> emitted.add(event)) + .ownerReference(true) + .build(); + + recorder.record(EventRecord.normal("Created", "created"), context(configMap())); + + assertThat(emitted.get(0).getMetadata().getOwnerReferences()) + .singleElement() + .satisfies( + owner -> { + assertThat(owner.getApiVersion()).isEqualTo("v1"); + assertThat(owner.getKind()).isEqualTo("ConfigMap"); + assertThat(owner.getName()).isEqualTo("test1"); + assertThat(owner.getUid()).isEqualTo("uid-1"); + }); + } + + @Test + void carriesNoOwnerReferenceByDefault() { + recorder.record(EventRecord.normal("Created", "created"), context(configMap())); + + assertThat(emitted.get(0).getMetadata().getOwnerReferences()).isEmpty(); + } + + @Test + void letsARecordOptOutOfTheRecorderLevelOwnerReference() { + var recorder = + DefaultEventRecorder.builder((event, context) -> emitted.add(event)) + .ownerReference(true) + .build(); + + recorder.record( + EventRecord.builder().reason("Created").message("created").ownedByRegarding(false).build(), + context(configMap())); + + assertThat(emitted.get(0).getMetadata().getOwnerReferences()).isEmpty(); + } + + @Test + void letsARecordOptIntoTheOwnerReferenceOnItsOwn() { + recorder.record( + EventRecord.builder().reason("Created").message("created").ownedByRegarding(true).build(), + context(configMap())); + + assertThat(emitted.get(0).getMetadata().getOwnerReferences()).hasSize(1); + } + + @Test + void setsNoOwnerReferenceWhenTheRegardingObjectHasNoUidYet() { + var withoutUid = configMap(); + withoutUid.getMetadata().setUid(null); + + recorder.record( + EventRecord.builder().reason("Created").message("created").ownedByRegarding(true).build(), + context(withoutUid)); + + assertThat(emitted.get(0).getMetadata().getOwnerReferences()).isEmpty(); + } + + @Test + void namesTheEventAfterThePerRecordNameWhenOneIsSet() { + recorder.record( + EventRecord.builder().reason("Created").message("created").name("my-event-name").build(), + context(configMap())); + + assertThat(emitted.get(0).getMetadata().getName()).isEqualTo("my-event-name"); + } + + @Test + void namesEventsThroughTheNamingStrategy() { + var recorder = + DefaultEventRecorder.builder((event, context) -> emitted.add(event)) + .namingStrategy( + (regarding, record) -> + Optional.of(regarding.getMetadata().getName() + "-" + record.reason())) + .build(); + + recorder.record(EventRecord.warning("failed", "first"), context(configMap())); + recorder.record(EventRecord.warning("failed", "second"), context(configMap())); + + assertThat(emitted) + .allSatisfy(event -> assertThat(event.getMetadata().getName()).isEqualTo("test1-failed")); + } + + @Test + void fallsBackToTheIdentityHashNameWhenTheNamingStrategyResolvesNothing() { + var recorder = + DefaultEventRecorder.builder((event, context) -> emitted.add(event)) + .namingStrategy((regarding, record) -> Optional.empty()) + .build(); + + recorder.record(EventRecord.warning("Failed", "message"), context(configMap())); + + assertThat(emitted.get(0).getMetadata().getName()) + .startsWith("test1.") + .hasSize("test1.".length() + 32); + } + + @Test + void truncatesSuppliedNamesToTheKubernetesNameLengthLimit() { + recorder.record( + EventRecord.builder().reason("Created").message("created").name("a".repeat(300)).build(), + context(configMap())); + + assertThat(emitted.get(0).getMetadata().getName()).hasSize(253); + } + + @Test + void stripsTrailingSeparatorsFromTruncatedNames() { + recorder.record( + EventRecord.builder() + .reason("Created") + .message("created") + .name("a".repeat(252) + "." + "b".repeat(47)) + .build(), + context(configMap())); + recorder.record( + EventRecord.builder() + .reason("Created") + .message("created") + .name("b".repeat(252) + "-" + "c".repeat(47)) + .build(), + context(configMap())); + + assertThat(emitted.get(0).getMetadata().getName()).isEqualTo("a".repeat(252)); + assertThat(emitted.get(1).getMetadata().getName()).isEqualTo("b".repeat(252)); + } + + @Test + void treatsABlankNameAsUnset() { + var recorder = + DefaultEventRecorder.builder((event, context) -> emitted.add(event)) + .namingStrategy((regarding, record) -> Optional.of("strategy-name")) + .build(); + + recorder.record( + EventRecord.builder().reason("Created").message("created").name("").build(), + context(configMap())); + + assertThat(emitted.get(0).getMetadata().getName()).isEqualTo("strategy-name"); + } + + @Test + void fallsBackToTheIdentityHashNameWhenEveryNameIsBlank() { + var recorder = + DefaultEventRecorder.builder((event, context) -> emitted.add(event)) + .namingStrategy((regarding, record) -> Optional.of(" ")) + .build(); + + recorder.record( + EventRecord.builder().reason("Created").message("created").name("").build(), + context(configMap())); + + assertThat(emitted.get(0).getMetadata().getName()).matches("test1\\.[0-9a-f]{32}"); + } + + @Test + void fallsBackToTheIdentityHashNameWhenTruncationLeavesNothing() { + recorder.record( + EventRecord.builder().reason("Created").message("created").name(".".repeat(300)).build(), + context(configMap())); + + assertThat(emitted.get(0).getMetadata().getName()).matches("test1\\.[0-9a-f]{32}"); + } + + @Test + void fallsBackToTheIdentityHashNameWhenTheSuppliedNameIsInvalid() { + recorder.record( + EventRecord.builder().reason("Created").message("created").name("Status").build(), + context(configMap())); + recorder.record( + EventRecord.builder().reason("Created").message("created").name("status_name").build(), + context(configMap())); + + assertThat(emitted) + .allSatisfy( + event -> assertThat(event.getMetadata().getName()).matches("test1\\.[0-9a-f]{32}")); + } + + @Test + void aFailingNamingStrategyNeverFailsTheCaller() { + var recorder = + DefaultEventRecorder.builder((event, context) -> emitted.add(event)) + .namingStrategy( + (regarding, record) -> { + throw new RuntimeException("cannot derive a name"); + }) + .build(); + + assertThatCode( + () -> recorder.record(EventRecord.normal("Created", "created"), context(configMap()))) + .doesNotThrowAnyException(); + assertThat(emitted).isEmpty(); + } + Context context(HasMetadata primaryResource) { return context(primaryResource, DefaultEventRecorder.CLUSTER_SCOPED_EVENT_NAMESPACE); } @@ -259,7 +528,7 @@ Context context(HasMetadata primaryResource) { * the configuration service the reporting instance and the cluster scoped event namespace come * from. */ - @SuppressWarnings({"unchecked", "rawtypes"}) + @SuppressWarnings("rawtypes") Context context(HasMetadata primaryResource, String clusterScopedEventNamespace) { var configurationService = mock(ConfigurationService.class); when(configurationService.getLeaderElectionConfiguration()) diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java index 1db69a1f9e..7c36aabe66 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/AbstractDependentResourceTest.java @@ -15,6 +15,7 @@ */ package io.javaoperatorsdk.operator.processing.dependent; +import java.util.List; import java.util.Optional; import java.util.Set; @@ -23,8 +24,12 @@ import io.fabric8.kubernetes.api.model.ConfigMap; import io.fabric8.kubernetes.api.model.ConfigMapBuilder; import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +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.DesiredStateAspect; +import io.javaoperatorsdk.operator.processing.Controller; import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; import static org.junit.jupiter.api.Assertions.*; @@ -37,7 +42,18 @@ class AbstractDependentResourceTest { private static final DefaultContext CONTEXT = createContext(PRIMARY); private static DefaultContext createContext(TestCustomResource primary) { - return new DefaultContext<>(mock(), mock(), primary, false, false); + return createContext(primary, List.of()); + } + + private static DefaultContext createContext( + TestCustomResource primary, 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 new DefaultContext<>(mock(), controller, primary, false, false); } @Test @@ -101,6 +117,35 @@ void checkThatDesiredIsOnlyCalledOnce() { assertEquals(1, testDependentResource.desiredCallCount); } + @Test + void appliesConfiguredDesiredStateAspectsInOrderAndOnlyOnce() { + final var testDependentResource = new DesiredCallCountCheckingDR(); + final var primary = new TestCustomResource(); + final var spec = primary.getSpec(); + spec.setConfigMapName("foo"); + spec.setKey("key"); + spec.setValue("value"); + final var context = + createContext( + primary, + List.of( + (desired, dependentResource, ctx) -> { + assertSame(testDependentResource, dependentResource); + assertSame(primary, ctx.getPrimaryResource()); + desired.getMetadata().getLabels().put("aspect", "first"); + }, + (desired, dependentResource, ctx) -> + desired.getMetadata().getLabels().put("aspect", "second"))); + + final var created = testDependentResource.reconcile(primary, context).getSingleResource(); + assertEquals("second", created.orElseThrow().getMetadata().getLabels().get("aspect")); + + // desired state is cached, aspects should therefore not be applied again + created.orElseThrow().getMetadata().getLabels().remove("aspect"); + testDependentResource.reconcile(primary, context); + assertNull(created.orElseThrow().getMetadata().getLabels().get("aspect")); + } + private ConfigMap configMap() { ConfigMap configMap = new ConfigMap(); configMap.setMetadata( diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcherTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcherTest.java index 0c2583d594..51bba5c76d 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcherTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/dependent/kubernetes/GenericKubernetesResourceMatcherTest.java @@ -15,6 +15,7 @@ */ package io.javaoperatorsdk.operator.processing.dependent.kubernetes; +import java.util.List; import java.util.Map; import java.util.Optional; @@ -27,12 +28,16 @@ import io.fabric8.kubernetes.client.KubernetesClient; import io.javaoperatorsdk.operator.MockKubernetesClient; import io.javaoperatorsdk.operator.ReconcilerUtilsInternal; +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.processing.Controller; import static io.javaoperatorsdk.operator.processing.dependent.kubernetes.GenericKubernetesResourceMatcher.match; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; @SuppressWarnings({"unchecked"}) class GenericKubernetesResourceMatcherTest { @@ -47,7 +52,18 @@ public TestContext() { } public TestContext(HasMetadata primary) { - super(mock(), mock(), primary, false, false); + super(mock(), mockController(), primary, false, false); + } + + @SuppressWarnings("rawtypes") + private static Controller mockController() { + final ConfigurationService configurationService = mock(); + when(configurationService.desiredStateAspects()).thenReturn(List.of()); + final ControllerConfiguration controllerConfiguration = mock(); + when(controllerConfiguration.getConfigurationService()).thenReturn(configurationService); + final Controller controller = mock(); + when(controller.getConfiguration()).thenReturn(controllerConfiguration); + return controller; } @Override diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManagerConcurrentReleaseTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManagerConcurrentReleaseTest.java index e8801781db..6fe3b3734e 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManagerConcurrentReleaseTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerManagerConcurrentReleaseTest.java @@ -52,7 +52,7 @@ * InformerManager#changeNamespaces(Set)} can run concurrently, so removing the source from the * manager has to be what claims the right to release it. */ -@SuppressWarnings({"rawtypes", "unchecked"}) +@SuppressWarnings("unchecked") class InformerManagerConcurrentReleaseTest { private static final String NAMESPACE = "ns1"; @@ -94,6 +94,9 @@ void concurrentStopAndNamespaceChangeReleaseTheInformerOnlyOnce() throws Excepti pool.proceed.countDown(); namespaceChange.join(TimeUnit.SECONDS.toMillis(5)); + assertThat(pool.proceededInTime) + .as("the blocked release must be let through, otherwise the paths never interleaved") + .isTrue(); assertThat(pool.releaseCount.get()) .as("the same namespace must not be released twice") .isEqualTo(1); @@ -114,7 +117,7 @@ private ControllerConfiguration controllerConfiguration() { /** Has to match what the manager builds for {@link #NAMESPACE} so it hits the same pool entry. */ private InformerClassifier classifier() { return new InformerClassifier<>( - clientMock, null, null, NAMESPACE, Deployment.class, null, null, null, null); + clientMock, null, null, NAMESPACE, Deployment.class, null, null, null, null, false); } /** Blocks inside the first release so the two teardown paths can be interleaved on purpose. */ @@ -124,6 +127,7 @@ private static class LatchingInformerPool extends DefaultInformerPool { private final CountDownLatch proceed = new CountDownLatch(1); private final AtomicInteger releaseCount = new AtomicInteger(); private final AtomicBoolean blockNextRelease = new AtomicBoolean(true); + private final AtomicBoolean proceededInTime = new AtomicBoolean(); @Override public Optional> releaseInformer( @@ -134,7 +138,7 @@ public Optional> releaseInformer( if (blockNextRelease.compareAndSet(true, false)) { enteredRelease.countDown(); try { - proceed.await(5, TimeUnit.SECONDS); + proceededInTime.set(proceed.await(5, TimeUnit.SECONDS)); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapperTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapperTest.java index e175074125..0bdcee35df 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapperTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/InformerWrapperTest.java @@ -112,7 +112,7 @@ private InformerWrapper wrapper(String controller, String ev informer, "default", new InformerClassifier<>( - null, null, null, "default", TestCustomResource.class, null, null, null, null), + null, null, null, "default", TestCustomResource.class, null, null, null, null, false), controller, eventSource); } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPoolTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPoolTest.java new file mode 100644 index 0000000000..e5a1208f06 --- /dev/null +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/AbstractInformerPoolTest.java @@ -0,0 +1,76 @@ +/* + * Copyright Java Operator SDK Authors + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package io.javaoperatorsdk.operator.processing.event.source.informer.pool; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import io.fabric8.kubernetes.client.KubernetesClient; +import io.javaoperatorsdk.operator.MockKubernetesClient; +import io.javaoperatorsdk.operator.api.config.BaseConfigurationService; +import io.javaoperatorsdk.operator.sample.simple.TestCustomResource; + +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +/** + * Unit tests for the informer creation that {@link AbstractInformerPool} performs for every pool, + * as opposed to the sharing strategy a concrete pool adds on top (covered by {@link + * DefaultInformerPoolTest} and {@link NonSharingInformerPoolTest}). {@link NonSharingInformerPool} + * is used merely as the simplest concrete subclass to reach that inherited behavior through. + */ +class AbstractInformerPoolTest { + + private static final String CONTROLLER = "controller"; + private static final String ES_NAME = "event-source"; + private static final String NAMESPACE = "default"; + + private final KubernetesClient client = MockKubernetesClient.client(TestCustomResource.class); + private final NonSharingInformerPool pool = new NonSharingInformerPool(); + + @BeforeEach + void setUp() { + pool.setConfigurationService(new BaseConfigurationService()); + } + + @Test + void keepsTheNamespaceIndexWhenTheClassifierDoesNotAskForIt() { + var informer = pool.getInformer(CONTROLLER, ES_NAME, classifier(false)); + + verify(informer, never()).removeNamespaceIndex(); + } + + @Test + void removesTheNamespaceIndexWhenTheClassifierAsksForIt() { + var informer = pool.getInformer(CONTROLLER, ES_NAME, classifier(true)); + + verify(informer).removeNamespaceIndex(); + } + + private InformerClassifier classifier(boolean withoutNamespaceIndex) { + return new InformerClassifier<>( + client, + null, + null, + NAMESPACE, + TestCustomResource.class, + null, + null, + null, + null, + withoutNamespaceIndex); + } +} diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPoolTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPoolTest.java index 4ca4db22bd..744dbdccee 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPoolTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/DefaultInformerPoolTest.java @@ -89,10 +89,10 @@ void createsSeparateInformersForDifferentClientsWithTheSameApiServerUrl() { void sharesInformerWhenClassifiersDifferOnlyByListLimit() { var withLimit100 = new InformerClassifier<>( - client, null, null, "default", TestCustomResource.class, null, null, 100L, null); + client, null, null, "default", TestCustomResource.class, null, null, 100L, null, false); var withLimit200 = new InformerClassifier<>( - client, null, null, "default", TestCustomResource.class, null, null, 200L, null); + client, null, null, "default", TestCustomResource.class, null, null, 200L, null, false); var first = pool.getInformer(CONTROLLER, ES_NAME, withLimit100); var second = pool.getInformer("other-controller", "other-es", withLimit200); @@ -149,6 +149,6 @@ private InformerClassifier classifier(String namespace) { private InformerClassifier classifier( KubernetesClient forClient, String namespace) { return new InformerClassifier<>( - forClient, null, null, namespace, TestCustomResource.class, null, null, null, null); + forClient, null, null, namespace, TestCustomResource.class, null, null, null, null, false); } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifierTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifierTest.java index 47f359a3f3..9025158918 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifierTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/InformerClassifierTest.java @@ -45,7 +45,7 @@ class InformerClassifierTest { private static final FieldSelector FIELD_SELECTOR = new FieldSelector(new FieldSelector.Field("status.phase", "Running")); private static final Long LIMIT = 100L; - private static final ItemStore ITEM_STORE = mock(ItemStore.class); + private static final ItemStore ITEM_STORE = mock(); private static InformerClassifier base() { return new InformerClassifier<>( @@ -57,7 +57,8 @@ private static InformerClassifier base() { GVK, FIELD_SELECTOR, LIMIT, - ITEM_STORE); + ITEM_STORE, + false); } @Test @@ -78,7 +79,8 @@ void informerListLimitIsExcludedFromEqualityAndHashCode() { GVK, FIELD_SELECTOR, 999L, - ITEM_STORE); + ITEM_STORE, + false); assertThat(base()).isEqualTo(withOtherLimit); assertThat(base()).hasSameHashCodeAs(withOtherLimit); @@ -103,7 +105,8 @@ void toStringContainsTheApiServerUrlDerivedFromTheClient() { GVK, FIELD_SELECTOR, LIMIT, - ITEM_STORE); + ITEM_STORE, + false); assertThat(classifier.toString()) .contains("https://localhost:8443/") @@ -124,7 +127,8 @@ void toStringDoesNotFailWithoutAClient() { GVK, FIELD_SELECTOR, LIMIT, - ITEM_STORE); + ITEM_STORE, + false); assertThat(classifier.toString()).contains(NAMESPACE); } @@ -144,7 +148,8 @@ void differsWhenClientDiffers() { GVK, FIELD_SELECTOR, LIMIT, - ITEM_STORE)); + ITEM_STORE, + false)); } @Test @@ -160,7 +165,8 @@ void differsWhenLabelSelectorDiffers() { GVK, FIELD_SELECTOR, LIMIT, - ITEM_STORE)); + ITEM_STORE, + false)); } @Test @@ -176,7 +182,8 @@ void differsWhenShardSelectorDiffers() { GVK, FIELD_SELECTOR, LIMIT, - ITEM_STORE)); + ITEM_STORE, + false)); } @Test @@ -192,14 +199,16 @@ void differsWhenNamespaceDiffers() { GVK, FIELD_SELECTOR, LIMIT, - ITEM_STORE)); + ITEM_STORE, + false)); } @Test void differsWhenResourceClassDiffers() { - // item stores are null here because their generic type is tied to the resource class, which is - // exactly the field under test; this keeps the resource class the only difference. - var forTestResource = + // the resource class is the field under test, so everything tied to it follows: the item stores + // are null to keep it the only difference, and both are declared as wildcards since their type + // argument differs with it + InformerClassifier forTestResource = new InformerClassifier<>( CLIENT, LABEL, @@ -209,8 +218,9 @@ void differsWhenResourceClassDiffers() { GVK, FIELD_SELECTOR, LIMIT, - null); - var forOtherResource = + null, + false); + InformerClassifier forOtherResource = new InformerClassifier<>( CLIENT, LABEL, @@ -220,7 +230,8 @@ void differsWhenResourceClassDiffers() { GVK, FIELD_SELECTOR, LIMIT, - null); + null, + false); assertThat(forTestResource).isNotEqualTo(forOtherResource); } @@ -238,7 +249,8 @@ void differsWhenGroupVersionKindDiffers() { new GroupVersionKind("sample.io/v1", "Bar"), FIELD_SELECTOR, LIMIT, - ITEM_STORE)); + ITEM_STORE, + false)); } @Test @@ -254,7 +266,8 @@ void differsWhenFieldSelectorDiffers() { GVK, new FieldSelector(new FieldSelector.Field("status.phase", "Pending")), LIMIT, - ITEM_STORE)); + ITEM_STORE, + false)); } @Test @@ -270,7 +283,25 @@ void differsWhenItemStoreDiffers() { GVK, FIELD_SELECTOR, LIMIT, - mock(ItemStore.class))); + mock(), + false)); + } + + @Test + void differsWhenWithoutNamespaceIndexDiffers() { + assertThat(base()) + .isNotEqualTo( + new InformerClassifier<>( + CLIENT, + LABEL, + SHARD, + NAMESPACE, + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + LIMIT, + ITEM_STORE, + true)); } @Test @@ -285,7 +316,8 @@ void differsOnlyByInformerListLimitIsTrueWhenOnlyLimitDiffers() { GVK, FIELD_SELECTOR, 999L, - ITEM_STORE); + ITEM_STORE, + false); assertThat(base().differsOnlyByInformerListLimit(withOtherLimit)).isTrue(); } @@ -308,8 +340,51 @@ void differsOnlyByInformerListLimitIsFalseWhenAnotherFieldDiffers() { GVK, FIELD_SELECTOR, 999L, - ITEM_STORE); + ITEM_STORE, + false); assertThat(base().differsOnlyByInformerListLimit(differentNamespaceAndLimit)).isFalse(); } + + @Test + void differsOnlyByNamespaceIndexIsTrueWhenOnlyIndexDiffers() { + var withoutIndex = + new InformerClassifier<>( + CLIENT, + LABEL, + SHARD, + NAMESPACE, + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + LIMIT, + ITEM_STORE, + true); + + assertThat(base().differsOnlyByNamespaceIndex(withoutIndex)).isTrue(); + } + + @Test + void differsOnlyByNamespaceIndexIsFalseWhenFullyEqual() { + assertThat(base().differsOnlyByNamespaceIndex(base())).isFalse(); + } + + @Test + void differsOnlyByNamespaceIndexIsFalseWhenAnotherFieldDiffers() { + // different namespace AND different index setting: not "only by namespace index" + var differentNamespaceAndIndex = + new InformerClassifier<>( + CLIENT, + LABEL, + SHARD, + "other-ns", + TestCustomResource.class, + GVK, + FIELD_SELECTOR, + LIMIT, + ITEM_STORE, + true); + + assertThat(base().differsOnlyByNamespaceIndex(differentNamespaceAndIndex)).isFalse(); + } } diff --git a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/NonSharingInformerPoolTest.java b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/NonSharingInformerPoolTest.java index 364103849e..939905f065 100644 --- a/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/NonSharingInformerPoolTest.java +++ b/operator-framework-core/src/test/java/io/javaoperatorsdk/operator/processing/event/source/informer/pool/NonSharingInformerPoolTest.java @@ -138,6 +138,6 @@ void releaseOfUnknownInformerReturnsEmptyAndDoesNotThrow() { private InformerClassifier classifier(String namespace) { return new InformerClassifier<>( - client, null, null, namespace, TestCustomResource.class, null, null, null, null); + client, null, null, namespace, TestCustomResource.class, null, null, null, null, false); } } diff --git a/operator-framework-junit/pom.xml b/operator-framework-junit/pom.xml index 4ddbb31b3e..aa18d5c778 100644 --- a/operator-framework-junit/pom.xml +++ b/operator-framework-junit/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.6.1-SNAPSHOT + 999-SNAPSHOT operator-framework-junit diff --git a/operator-framework/pom.xml b/operator-framework/pom.xml index 4f57216ed4..6d314d4687 100644 --- a/operator-framework/pom.xml +++ b/operator-framework/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.6.1-SNAPSHOT + 999-SNAPSHOT operator-framework 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..f15ffaf78b 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, @@ -170,7 +174,11 @@ public static ConfigLoader getDefault() { new ConfigBinding<>( "informer.list-limit", Long.class, - ControllerConfigurationOverrider::withInformerListLimit)); + ControllerConfigurationOverrider::withInformerListLimit), + new ConfigBinding<>( + "informer.without-namespace-index", + Boolean.class, + ControllerConfigurationOverrider::withoutNamespaceIndex)); private final ConfigProvider configProvider; @@ -228,7 +236,7 @@ Consumer> applyControllerConfigs(String cont Consumer> retryStep = buildRetryConsumer(prefix); if (retryStep != null) { - consumer = consumer == null ? retryStep : consumer.andThen(retryStep); + consumer = consumer.andThen(retryStep); } Consumer> rateLimiterStep = buildRateLimiterConsumer(prefix); diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/eventrecorderconfigured/ConfiguredEventRecorderCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/eventrecorderconfigured/ConfiguredEventRecorderCustomResource.java new file mode 100644 index 0000000000..0e0e3cab5d --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/eventrecorderconfigured/ConfiguredEventRecorderCustomResource.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.eventrecorderconfigured; + +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("ConfiguredEventRecorderCustomResource") +@ShortNames("cerc") +public class ConfiguredEventRecorderCustomResource extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/eventrecorderconfigured/ConfiguredEventRecorderIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/eventrecorderconfigured/ConfiguredEventRecorderIT.java new file mode 100644 index 0000000000..537aa38f8d --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/eventrecorderconfigured/ConfiguredEventRecorderIT.java @@ -0,0 +1,168 @@ +/* + * 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.eventrecorderconfigured; + +import java.util.List; +import java.util.Map; +import java.util.Optional; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.Event; +import io.fabric8.kubernetes.api.model.HasMetadata; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.OwnerReference; +import io.fabric8.kubernetes.client.KubernetesClient; +import io.fabric8.kubernetes.client.KubernetesClientBuilder; +import io.javaoperatorsdk.annotation.Sample; +import io.javaoperatorsdk.operator.api.event.DefaultEventRecorder; +import io.javaoperatorsdk.operator.api.event.DefaultEventSink; +import io.javaoperatorsdk.operator.api.event.EventKeyStrategy; +import io.javaoperatorsdk.operator.api.event.EventRecord; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static io.javaoperatorsdk.operator.baseapi.eventrecorderconfigured.ConfiguredEventRecorderReconciler.AGGREGATED_REASON; +import static io.javaoperatorsdk.operator.baseapi.eventrecorderconfigured.ConfiguredEventRecorderReconciler.NAMED_REASON; +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +@Sample( + tldr = "Configuring how the event recorder aggregates, names and owns events", + description = + """ + Demonstrates configuring the event recorder with a default aggregation key strategy, a \ + naming strategy and owner references. Aggregating by reason resolves repeated occurrences \ + with changing messages to one event whose count grows and whose message is replaced with \ + the latest one. A naming strategy gives events predictable names instead of the default \ + identity hash. Owner references relate the events to the object they are about. + """) +class ConfiguredEventRecorderIT { + + private static final String TEST_RESOURCE_NAME = "test1"; + + private final KubernetesClient sinkClient = new KubernetesClientBuilder().build(); + + private final ConfiguredEventRecorderReconciler reconciler = + new ConfiguredEventRecorderReconciler(); + + @RegisterExtension + LocallyRunOperatorExtension extension = + LocallyRunOperatorExtension.builder() + .withReconciler(reconciler) + .withConfigurationService(o -> o.withEventRecorder(configuredEventRecorder())) + .build(); + + @AfterEach + void closeSinkClient() { + sinkClient.close(); + } + + @Test + void aggregatesOccurrencesWithChangingMessagesOntoOneEvent() { + var resource = extension.create(testResource()); + await().untilAsserted(() -> assertThat(reconciler.getNumberOfExecutions()).isPositive()); + + // reconcile once more: aggregating by reason resolves the new messages to the same events + resource.getMetadata().setAnnotations(Map.of("reconcile", "again")); + extension.replace(resource); + + await() + .untilAsserted( + () -> { + assertThat(reconciler.getNumberOfExecutions()).isGreaterThanOrEqualTo(2); + + assertThat(eventsWithReason(AGGREGATED_REASON)) + .singleElement() + .satisfies( + event -> { + assertThat(event.getCount()).isGreaterThanOrEqualTo(2); + assertThat(event.getMessage()) + .isEqualTo("something changed in execution " + event.getCount()); + }); + }); + } + + @Test + void namesEventsThroughTheConfiguredNamingStrategy() { + extension.create(testResource()); + + await() + .untilAsserted( + () -> + assertThat(eventsWithReason(NAMED_REASON)) + .singleElement() + .extracting(event -> event.getMetadata().getName()) + .isEqualTo(TEST_RESOURCE_NAME + "-status-report")); + } + + @Test + void setsOwnerReferencesOnTheEventsItRecords() { + var resource = extension.create(testResource()); + + await() + .untilAsserted( + () -> { + var events = eventsForTestResource(); + assertThat(events).isNotEmpty(); + assertThat(events) + .allSatisfy( + event -> + assertThat(event.getMetadata().getOwnerReferences()) + .singleElement() + .returns( + "ConfiguredEventRecorderCustomResource", OwnerReference::getKind) + .returns(resource.getMetadata().getUid(), OwnerReference::getUid)); + }); + } + + private List eventsWithReason(String reason) { + return eventsForTestResource().stream().filter(e -> reason.equals(e.getReason())).toList(); + } + + @SuppressWarnings("resource") + private List eventsForTestResource() { + return extension + .getKubernetesClient() + .v1() + .events() + .inNamespace(extension.getNamespace()) + .withField("involvedObject.name", TEST_RESOURCE_NAME) + .list() + .getItems(); + } + + private ConfiguredEventRecorderCustomResource testResource() { + var resource = new ConfiguredEventRecorderCustomResource(); + resource.setMetadata(new ObjectMetaBuilder().withName(TEST_RESOURCE_NAME).build()); + return resource; + } + + private DefaultEventRecorder configuredEventRecorder() { + return DefaultEventRecorder.builder(new DefaultEventSink(sinkClient)) + .namingStrategy(ConfiguredEventRecorderIT::statusReportName) + .keyStrategy(EventKeyStrategy.byReason()) + .ownerReference(true) + .build(); + } + + private static Optional statusReportName(HasMetadata regarding, EventRecord record) { + return NAMED_REASON.equals(record.reason()) + ? Optional.of(regarding.getMetadata().getName() + "-status-report") + : Optional.empty(); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/eventrecorderconfigured/ConfiguredEventRecorderReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/eventrecorderconfigured/ConfiguredEventRecorderReconciler.java new file mode 100644 index 0000000000..af4d136a4d --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/eventrecorderconfigured/ConfiguredEventRecorderReconciler.java @@ -0,0 +1,49 @@ +/* + * 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.eventrecorderconfigured; + +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; + +@ControllerConfiguration(generationAwareEventProcessing = false) +public class ConfiguredEventRecorderReconciler + implements Reconciler { + + public static final String NAMED_REASON = "StatusReport"; + public static final String AGGREGATED_REASON = "SomethingChanged"; + + private final AtomicInteger numberOfExecutions = new AtomicInteger(); + + @Override + public UpdateControl reconcile( + ConfiguredEventRecorderCustomResource resource, + Context context) { + var execution = numberOfExecutions.incrementAndGet(); + context.eventRecorder().warn(NAMED_REASON, "status after execution " + execution); + context + .eventRecorder() + .normal(AGGREGATED_REASON, "something changed in execution " + execution); + return UpdateControl.noUpdate(); + } + + public int getNumberOfExecutions() { + return numberOfExecutions.get(); + } +} 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/baseapi/withoutnamespaceindex/WithoutNamespaceIndexIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/withoutnamespaceindex/WithoutNamespaceIndexIT.java new file mode 100644 index 0000000000..bd68813608 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/withoutnamespaceindex/WithoutNamespaceIndexIT.java @@ -0,0 +1,76 @@ +/* + * 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.withoutnamespaceindex; + +import java.time.Duration; +import java.util.Map; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.ConfigMapBuilder; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.fabric8.kubernetes.api.model.SecretBuilder; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +class WithoutNamespaceIndexIT { + + private static final String TEST_RESOURCE_NAME = "test1"; + + @RegisterExtension + LocallyRunOperatorExtension extension = + LocallyRunOperatorExtension.builder() + .withReconciler(new WithoutNamespaceIndexTestReconciler()) + .build(); + + @Test + void reconcilesAndResolvesSecondariesWithoutTheNamespaceIndex() { + var configMap = + extension.create( + new ConfigMapBuilder() + .withMetadata(new ObjectMetaBuilder().withName(TEST_RESOURCE_NAME).build()) + .withData(Map.of("key", "value")) + .build()); + extension.create( + new SecretBuilder() + .withMetadata(new ObjectMetaBuilder().withName(TEST_RESOURCE_NAME).build()) + .build()); + + var reconciler = extension.getReconcilerOfType(WithoutNamespaceIndexTestReconciler.class); + await() + .pollDelay(Duration.ofMillis(150)) + .untilAsserted( + () -> { + assertThat(reconciler.getNumberOfExecutions()).isPositive(); + assertThat(reconciler.getSecondariesFound()).contains(TEST_RESOURCE_NAME); + }); + + // an update has to reach the reconciler too: the informer keeps serving events and cache reads + // with the index gone, it is not just the initial sync that works + var executionsBeforeUpdate = reconciler.getNumberOfExecutions(); + configMap.setData(Map.of("key", "updated")); + extension.update(configMap); + + await() + .untilAsserted( + () -> + assertThat(reconciler.getNumberOfExecutions()) + .isGreaterThan(executionsBeforeUpdate)); + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/withoutnamespaceindex/WithoutNamespaceIndexTestReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/withoutnamespaceindex/WithoutNamespaceIndexTestReconciler.java new file mode 100644 index 0000000000..930e41480a --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/baseapi/withoutnamespaceindex/WithoutNamespaceIndexTestReconciler.java @@ -0,0 +1,86 @@ +/* + * 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.withoutnamespaceindex; + +import java.util.Collections; +import java.util.HashSet; +import java.util.List; +import java.util.Set; +import java.util.concurrent.atomic.AtomicInteger; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.Secret; +import io.javaoperatorsdk.operator.api.config.informer.Informer; +import io.javaoperatorsdk.operator.api.config.informer.InformerEventSourceConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.Context; +import io.javaoperatorsdk.operator.api.reconciler.ControllerConfiguration; +import io.javaoperatorsdk.operator.api.reconciler.EventSourceContext; +import io.javaoperatorsdk.operator.api.reconciler.Reconciler; +import io.javaoperatorsdk.operator.api.reconciler.UpdateControl; +import io.javaoperatorsdk.operator.processing.event.ResourceID; +import io.javaoperatorsdk.operator.processing.event.source.EventSource; +import io.javaoperatorsdk.operator.processing.event.source.informer.InformerEventSource; +import io.javaoperatorsdk.operator.support.TestExecutionInfoProvider; + +/** + * Both informers drop the namespace index: the primary one through the annotation, the secondary + * one through the builder. Reconciling and resolving the secondary resource have to keep working, + * since neither reads that index. + */ +@ControllerConfiguration(informer = @Informer(withoutNamespaceIndex = true)) +public class WithoutNamespaceIndexTestReconciler + implements Reconciler, TestExecutionInfoProvider { + + private final AtomicInteger numberOfExecutions = new AtomicInteger(0); + private final Set secondariesFound = Collections.synchronizedSet(new HashSet<>()); + + @Override + public UpdateControl reconcile(Secret resource, Context context) { + numberOfExecutions.addAndGet(1); + // reads through the secondary informer's cache, the part that would break if the framework + // relied on the index it just removed + context + .getSecondaryResource(ConfigMap.class) + .ifPresent(configMap -> secondariesFound.add(configMap.getMetadata().getName())); + return UpdateControl.noUpdate(); + } + + @Override + public List> prepareEventSources(EventSourceContext context) { + return List.of( + new InformerEventSource<>( + InformerEventSourceConfiguration.from(ConfigMap.class, Secret.class) + .withNamespacesInheritedFromController() + .withoutNamespaceIndex(true) + // the ConfigMap of a Secret is the one sharing its name + .withSecondaryToPrimaryMapper( + configMap -> + Set.of( + new ResourceID( + configMap.getMetadata().getName(), + configMap.getMetadata().getNamespace()))) + .build())); + } + + @Override + public int getNumberOfExecutions() { + return numberOfExecutions.get(); + } + + public Set getSecondariesFound() { + return secondariesFound; + } +} 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..e604cdac35 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 @@ -26,6 +26,7 @@ import org.junit.jupiter.api.Test; +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.ConfigurationServiceOverrider; @@ -93,6 +94,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 +107,7 @@ void applyConfigsAppliesBooleanFlags() { assertThat(result.ssaBasedCreateUpdateMatchForDependentResources()).isFalse(); assertThat(result.useSSAToPatchPrimaryResource()).isFalse(); assertThat(result.cloneSecondaryResourcesWhenGettingFromCache()).isTrue(); + assertThat(result.useVirtualThreads()).isTrue(); } @Test @@ -218,10 +221,23 @@ public Optional getValue(String key, Class type) { "josdk.controller.ctrl.informer.label-selector", "josdk.controller.ctrl.informer.shard-selector", "josdk.controller.ctrl.informer.list-limit", + "josdk.controller.ctrl.informer.without-namespace-index", "josdk.controller.ctrl.rate-limiter.refresh-period", "josdk.controller.ctrl.rate-limiter.limit-for-period"); } + @Test + void applyControllerConfigsAppliesInformerWithoutNamespaceIndex() { + var loader = + new ConfigLoader( + mapProvider(Map.of("josdk.controller.ctrl.informer.without-namespace-index", true))); + var overrider = ControllerConfigurationOverrider.override(baseControllerConfig()); + + loader.applyControllerConfigs("ctrl").accept(overrider); + + assertThat(overrider.build().getInformerConfig().isWithoutNamespaceIndex()).isTrue(); + } + @Test void operatorKeyPrefixIsJosdkDot() { assertThat(ConfigLoader.DEFAULT_OPERATOR_KEY_PREFIX).isEqualTo("josdk."); @@ -567,7 +583,6 @@ private static boolean isTypeCompatible(Class methodParam, Class bindingTy if (methodParam == long.class && bindingType == Long.class) return true; if (methodParam == Long.class && bindingType == long.class) return true; if (methodParam == double.class && bindingType == Double.class) return true; - if (methodParam == Double.class && bindingType == double.class) return true; - return false; + return methodParam == Double.class && bindingType == double.class; } } diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/desiredstateaspect/DesiredStateAspectCustomResource.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/desiredstateaspect/DesiredStateAspectCustomResource.java new file mode 100644 index 0000000000..e5be695160 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/desiredstateaspect/DesiredStateAspectCustomResource.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.dependent.desiredstateaspect; + +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 DesiredStateAspectCustomResource extends CustomResource + implements Namespaced {} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/desiredstateaspect/DesiredStateAspectIT.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/desiredstateaspect/DesiredStateAspectIT.java new file mode 100644 index 0000000000..ef8048b47d --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/desiredstateaspect/DesiredStateAspectIT.java @@ -0,0 +1,117 @@ +/* + * 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.dependent.desiredstateaspect; + +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +import io.javaoperatorsdk.annotation.Sample; +import io.javaoperatorsdk.operator.junit.LocallyRunOperatorExtension; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +@Sample( + tldr = "Common metadata on all managed resources using desired state aspects", + description = + """ + Demonstrates how to register a global DesiredStateAspect on the ConfigurationService in \ + order to add common metadata, here labels identifying the operator and the dependent \ + resource the secondary resource originates from, to every Kubernetes resource managed by \ + the operator. Aspects are applied to the desired state right after it is computed, so the \ + added metadata is also taken into account when matching the actual resource against its \ + desired state. + """) +class DesiredStateAspectIT { + + public static final String TEST_RESOURCE_NAME = "test1"; + public static final String MANAGED_BY_LABEL_KEY = "app.kubernetes.io/managed-by"; + public static final String MANAGED_BY_LABEL_VALUE = "desired-state-aspect-operator"; + public static final String DEPENDENT_LABEL_KEY = "javaoperatorsdk.io/dependent"; + public static final String DEPENDENT_LABEL_VALUE = + DesiredStateAspectReconciler.ConfigMapDependentResource.class.getSimpleName().toLowerCase(); + + @RegisterExtension + LocallyRunOperatorExtension operator = + LocallyRunOperatorExtension.builder() + .withReconciler(DesiredStateAspectReconciler.class) + .withConfigurationService( + o -> + o.addDesiredStateAspects( + (desired, dependentResource, context) -> + desired + .getMetadata() + .getLabels() + .put(MANAGED_BY_LABEL_KEY, MANAGED_BY_LABEL_VALUE), + (desired, dependentResource, context) -> + desired + .getMetadata() + .getLabels() + .put( + DEPENDENT_LABEL_KEY, + dependentResource.getClass().getSimpleName().toLowerCase()))) + .build(); + + @Test + void aspectsAreAppliedToAllManagedResources() { + operator.create(testResource()); + + await() + .untilAsserted( + () -> { + var configMap = operator.get(ConfigMap.class, TEST_RESOURCE_NAME); + assertThat(configMap).isNotNull(); + assertThat(configMap.getMetadata().getLabels()) + .containsEntry(MANAGED_BY_LABEL_KEY, MANAGED_BY_LABEL_VALUE) + .containsEntry(DEPENDENT_LABEL_KEY, DEPENDENT_LABEL_VALUE); + }); + } + + @Test + void metadataAddedByAspectsIsRestoredIfRemoved() { + operator.create(testResource()); + + await() + .untilAsserted( + () -> + assertThat(operator.get(ConfigMap.class, TEST_RESOURCE_NAME)) + .isNotNull() + .extracting(cm -> cm.getMetadata().getLabels()) + .satisfies( + labels -> + assertThat(labels) + .containsEntry(MANAGED_BY_LABEL_KEY, MANAGED_BY_LABEL_VALUE))); + + var configMap = operator.get(ConfigMap.class, TEST_RESOURCE_NAME); + configMap.getMetadata().getLabels().remove(MANAGED_BY_LABEL_KEY); + operator.replace(configMap); + + await() + .untilAsserted( + () -> + assertThat( + operator.get(ConfigMap.class, TEST_RESOURCE_NAME).getMetadata().getLabels()) + .containsEntry(MANAGED_BY_LABEL_KEY, MANAGED_BY_LABEL_VALUE)); + } + + DesiredStateAspectCustomResource testResource() { + var res = new DesiredStateAspectCustomResource(); + res.setMetadata(new ObjectMetaBuilder().withName(TEST_RESOURCE_NAME).build()); + return res; + } +} diff --git a/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/desiredstateaspect/DesiredStateAspectReconciler.java b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/desiredstateaspect/DesiredStateAspectReconciler.java new file mode 100644 index 0000000000..bdb34435e3 --- /dev/null +++ b/operator-framework/src/test/java/io/javaoperatorsdk/operator/dependent/desiredstateaspect/DesiredStateAspectReconciler.java @@ -0,0 +1,59 @@ +/* + * 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.dependent.desiredstateaspect; + +import java.util.Map; + +import io.fabric8.kubernetes.api.model.ConfigMap; +import io.fabric8.kubernetes.api.model.ObjectMetaBuilder; +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.api.reconciler.Workflow; +import io.javaoperatorsdk.operator.api.reconciler.dependent.Dependent; +import io.javaoperatorsdk.operator.processing.dependent.kubernetes.CRUDKubernetesDependentResource; + +@Workflow( + dependents = @Dependent(type = DesiredStateAspectReconciler.ConfigMapDependentResource.class)) +@ControllerConfiguration +public class DesiredStateAspectReconciler implements Reconciler { + + @Override + public UpdateControl reconcile( + DesiredStateAspectCustomResource resource, + Context context) { + return UpdateControl.noUpdate(); + } + + public static class ConfigMapDependentResource + extends CRUDKubernetesDependentResource { + + @Override + protected ConfigMap desired( + DesiredStateAspectCustomResource primary, + Context context) { + ConfigMap configMap = new ConfigMap(); + configMap.setMetadata( + new ObjectMetaBuilder() + .withName(primary.getMetadata().getName()) + .withNamespace(primary.getMetadata().getNamespace()) + .build()); + configMap.setData(Map.of("data", primary.getMetadata().getName())); + return configMap; + } + } +} diff --git a/pom.xml b/pom.xml index 0aaad2e670..c88d2f4588 100644 --- a/pom.xml +++ b/pom.xml @@ -21,7 +21,7 @@ io.javaoperatorsdk java-operator-sdk - 5.6.1-SNAPSHOT + 999-SNAPSHOT pom Operator SDK for Java Java SDK for implementing Kubernetes operators diff --git a/sample-operators/controller-namespace-deletion/pom.xml b/sample-operators/controller-namespace-deletion/pom.xml index 33cf5ab823..af4be01972 100644 --- a/sample-operators/controller-namespace-deletion/pom.xml +++ b/sample-operators/controller-namespace-deletion/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.6.1-SNAPSHOT + 999-SNAPSHOT sample-controller-namespace-deletion diff --git a/sample-operators/kotlin-operator/pom.xml b/sample-operators/kotlin-operator/pom.xml index 70e5e99d5d..a5ca180cd2 100644 --- a/sample-operators/kotlin-operator/pom.xml +++ b/sample-operators/kotlin-operator/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.6.1-SNAPSHOT + 999-SNAPSHOT sample-kotlin-operator diff --git a/sample-operators/leader-election/pom.xml b/sample-operators/leader-election/pom.xml index 35cc48ca7f..4f896485d1 100644 --- a/sample-operators/leader-election/pom.xml +++ b/sample-operators/leader-election/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.6.1-SNAPSHOT + 999-SNAPSHOT sample-leader-election diff --git a/sample-operators/mysql-schema/pom.xml b/sample-operators/mysql-schema/pom.xml index a7cb233772..d2872c921a 100644 --- a/sample-operators/mysql-schema/pom.xml +++ b/sample-operators/mysql-schema/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.6.1-SNAPSHOT + 999-SNAPSHOT sample-mysql-schema-operator diff --git a/sample-operators/operations/pom.xml b/sample-operators/operations/pom.xml index 37f5a09fdc..239b8a4860 100644 --- a/sample-operators/operations/pom.xml +++ b/sample-operators/operations/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.6.1-SNAPSHOT + 999-SNAPSHOT sample-operations diff --git a/sample-operators/pom.xml b/sample-operators/pom.xml index 0ba6238e45..704007c076 100644 --- a/sample-operators/pom.xml +++ b/sample-operators/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk java-operator-sdk - 5.6.1-SNAPSHOT + 999-SNAPSHOT sample-operators diff --git a/sample-operators/tomcat-operator/pom.xml b/sample-operators/tomcat-operator/pom.xml index e7d8750a90..ea964a2b07 100644 --- a/sample-operators/tomcat-operator/pom.xml +++ b/sample-operators/tomcat-operator/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.6.1-SNAPSHOT + 999-SNAPSHOT sample-tomcat-operator diff --git a/sample-operators/webpage/pom.xml b/sample-operators/webpage/pom.xml index 3ef0f2a11f..d50e5ef03c 100644 --- a/sample-operators/webpage/pom.xml +++ b/sample-operators/webpage/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk sample-operators - 5.6.1-SNAPSHOT + 999-SNAPSHOT sample-webpage-operator 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/test-index-processor/pom.xml b/test-index-processor/pom.xml index b930d1de7d..2ae7c5f454 100644 --- a/test-index-processor/pom.xml +++ b/test-index-processor/pom.xml @@ -22,7 +22,7 @@ io.javaoperatorsdk java-operator-sdk - 5.6.1-SNAPSHOT + 999-SNAPSHOT test-index-processor