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.javaoperatorsdkjava-operator-sdk
- 5.6.1-SNAPSHOT
+ 999-SNAPSHOTbootstrapper
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.javaoperatorsdkjava-operator-sdk
- 5.6.1-SNAPSHOT
+ 999-SNAPSHOTcaffeine-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..aec261bbf0 100644
--- a/docs/content/en/docs/documentation/eventing.md
+++ b/docs/content/en/docs/documentation/eventing.md
@@ -222,6 +222,31 @@ is similar to `PerResourcePollingEventSource` except that, contrary to that even
doesn't poll a specific API separately per resource, but periodically and independently of
actually observed primary resources.
+#### Threading of the polling event sources
+
+Both polling event sources schedule their polls on an executor the operator shares between all of
+its polling event sources. A poll therefore only starts once one of that executor's threads is
+free, so if your operator registers many polling event sources, or if fetching your external
+resources is slow, size the pool accordingly with
+`ConfigurationServiceOverrider.withConcurrentScheduledTaskThreads` (4 threads by default):
+
+```java
+Operator operator = new Operator(overrider -> overrider.withConcurrentScheduledTaskThreads(20));
+```
+
+Retried and rescheduled reconciliations are triggered on a separate executor, so a slow poll can
+never delay them. It is sized with `withConcurrentRetryAndRescheduleThreads` (2 threads by
+default); few threads are needed there since triggering a reconciliation only enqueues an event
+for one of the reconciliation threads to pick up.
+
+Use `ConfigurationServiceOverrider.withScheduledExecutorService` to replace the polling executor
+altogether. The operator still manages the lifecycle of that executor: it shuts it down when the
+operator is stopped, so don't shut it down yourself.
+
+Use the `withExecutorService` method of the event source's own configuration builder to poll a
+single event source on an executor of its own. An executor provided that way is not managed by the
+operator: it is your responsibility to shut it down.
+
#### Inbound event sources
[SimpleInboundEventSource](https://github.com/operator-framework/java-operator-sdk/blob/main/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/processing/event/source/inbound/SimpleInboundEventSource.java)
@@ -348,6 +373,30 @@ See
also [CaffeineBoundedItemStores](https://github.com/operator-framework/java-operator-sdk/blob/main/caffeine-bounded-cache-support/src/main/java/io/javaoperatorsdk/operator/processing/event/source/cache/CaffeineBoundedItemStores.java)
for more details.
+### Removing the Namespace Index
+
+Informers keep an index from namespace to the resources cached for it. JOSDK never reads that index,
+only the ones registered explicitly through `IndexerResourceCache.addIndexers(..)`, so it can be
+removed to save an entry per cached resource:
+
+```java
+@ControllerConfiguration(informer = @Informer(withoutNamespaceIndex = true))
+public class MyReconciler implements Reconciler { }
+```
+
+The same option is available on `InformerEventSourceConfiguration.Builder` for event sources, and on
+`InformerConfiguration.Builder`.
+
+This matters most for informers that cache a large number of resources, and in particular together
+with a custom [item store](#bounded-caches-for-informers) that keeps only a reduced form of each
+resource: the index is keyed independently of what the store does with the resource itself, so
+shrinking what is cached does not shrink the index.
+
+Note that this takes part in the informer pool identity described below: an event source that
+removes the index does not share an informer with one that keeps it. If two event sources watch the
+same resource type and only one of them sets the option, they end up with two informers, and two
+caches of that resource type, which can cost far more memory than the index ever did.
+
### Sharing Informers Between Controllers (Informer Pool)
{{% alert title="Experimental" color="warning" %}}
@@ -374,7 +423,10 @@ Two event sources share an informer when their effective informer configuration
- the resource type (or the group/version/kind for generic resources),
- the watched namespace,
- the label, field and shard selectors,
-- the configured [item store](#bounded-caches-for-informers).
+- the configured [item store](#bounded-caches-for-informers),
+- whether the [namespace index is removed](#removing-the-namespace-index): it cannot be present for
+ one event source and absent for another on one shared informer, so event sources that disagree on
+ it are backed by separate informers.
The `informerListLimit` is intentionally *not* part of this identity: if two otherwise-equivalent
event sources request a different list limit, the existing informer is reused (a warning is logged
diff --git a/docs/content/en/docs/documentation/operations/configuration.md b/docs/content/en/docs/documentation/operations/configuration.md
index cdfb1b7fdb..eef800d66b 100644
--- a/docs/content/en/docs/documentation/operations/configuration.md
+++ b/docs/content/en/docs/documentation/operations/configuration.md
@@ -281,6 +281,13 @@ All operator-level keys are prefixed with `josdk.`.
|---|---|---|
| `josdk.workflow.executor-threads` | `Integer` | Thread pool size for workflow execution |
+#### Scheduled Tasks
+
+| Key | Type | Description |
+|---|---|---|
+| `josdk.scheduled-tasks.concurrent-threads` | `Integer` | Thread pool size for the operator's scheduled tasks, i.e. the polling event sources |
+| `josdk.retry-and-reschedule.concurrent-threads` | `Integer` | Thread pool size for triggering retried and rescheduled reconciliations |
+
#### Informer
| Key | Type | Description |
@@ -356,6 +363,7 @@ All controller-level keys are prefixed with `josdk.controller..
| `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.javaoperatorsdkjava-operator-sdk
- 5.6.1-SNAPSHOT
+ 999-SNAPSHOTmicrometer-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.javaoperatorsdkjava-operator-sdk
- 5.6.1-SNAPSHOT
+ 999-SNAPSHOTmigration
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.javaoperatorsdkoperator-framework-bom
- 5.6.1-SNAPSHOT
+ 999-SNAPSHOTpomOperator SDK - Bill of MaterialsJava 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.javaoperatorsdkjava-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..2dec5aabf2 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,12 @@
package io.javaoperatorsdk.operator.api.config;
import java.time.Duration;
+import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Consumer;
import org.slf4j.Logger;
@@ -41,6 +43,7 @@
import io.javaoperatorsdk.operator.api.reconciler.Experimental;
import io.javaoperatorsdk.operator.api.reconciler.Reconciler;
import io.javaoperatorsdk.operator.api.reconciler.dependent.DependentResourceFactory;
+import io.javaoperatorsdk.operator.api.reconciler.dependent.DesiredStateAspect;
import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependent;
import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResource;
import io.javaoperatorsdk.operator.processing.dependent.kubernetes.KubernetesDependentResourceConfig;
@@ -62,6 +65,20 @@ public interface ConfigurationService {
/** The default number of threads used to process dependent workflows */
int DEFAULT_WORKFLOW_EXECUTOR_THREAD_NUMBER = DEFAULT_RECONCILIATION_THREADS_NUMBER;
+ /**
+ * The default number of threads used to run the operator's scheduled tasks, i.e. the periodic
+ * polls of {@link io.javaoperatorsdk.operator.processing.event.source.polling.PollingEventSource}
+ * and {@link
+ * io.javaoperatorsdk.operator.processing.event.source.polling.PerResourcePollingEventSource}
+ */
+ int DEFAULT_SCHEDULED_TASK_THREADS_NUMBER = 4;
+
+ /**
+ * The default number of threads used to trigger the operator's retried and rescheduled
+ * reconciliations
+ */
+ int DEFAULT_RETRY_AND_RESCHEDULE_THREADS_NUMBER = 2;
+
/**
* Creates a new {@link ConfigurationService} instance used to configure an {@link
* io.javaoperatorsdk.operator.Operator} instance, starting from the specified base configuration
@@ -217,6 +234,33 @@ default int concurrentWorkflowExecutorThreads() {
return DEFAULT_WORKFLOW_EXECUTOR_THREAD_NUMBER;
}
+ /**
+ * Number of threads the operator can spin out to run its scheduled (i.e. periodic or delayed)
+ * tasks with the default executor. These threads are shared by all the polling event sources of
+ * the operator, so this number should be raised when many, or slow, polling event sources are
+ * registered: a poll only starts once a thread is available, and a slow poll therefore delays the
+ * polls of the other event sources.
+ *
+ * @return the maximum number of concurrent scheduled task threads
+ * @since 5.6.0
+ */
+ default int concurrentScheduledTaskThreads() {
+ return DEFAULT_SCHEDULED_TASK_THREADS_NUMBER;
+ }
+
+ /**
+ * Number of threads the operator can spin out to trigger its retried and rescheduled
+ * reconciliations with the default executor. These threads are shared by all the controllers of
+ * the operator, but the tasks they run only enqueue an event for the reconciliation to happen on
+ * a reconciliation thread, so few of them are needed.
+ *
+ * @return the maximum number of concurrent retry and reschedule threads
+ * @since 5.6.0
+ */
+ default int concurrentRetryAndRescheduleThreads() {
+ return DEFAULT_RETRY_AND_RESCHEDULE_THREADS_NUMBER;
+ }
+
/**
* Override to provide a custom {@link Metrics} implementation
*
@@ -247,6 +291,44 @@ default ExecutorService getWorkflowExecutorService() {
return Executors.newFixedThreadPool(concurrentWorkflowExecutorThreads());
}
+ /**
+ * Override to provide a custom {@link ScheduledExecutorService} implementation to change how the
+ * operator's scheduled (i.e. periodic or delayed) tasks are run. This executor is shared by all
+ * the polling event sources of the operator. The retried and rescheduled reconciliations run on
+ * an executor of their own, see {@link #getRetryAndRescheduleExecutorService()}, so that a slow
+ * poll can't delay them.
+ *
+ *
Note that the default implementation lets the executor discard the tasks that were scheduled
+ * for later when it is shut down, so that they don't delay the termination of the operator, and
+ * that it creates daemon threads so that a never stopped operator doesn't keep the JVM alive.
+ * Custom implementations are advised to do the same.
+ *
+ * @return the {@link ScheduledExecutorService} implementation to use to run scheduled tasks
+ * @since 5.6.0
+ */
+ default ScheduledExecutorService getScheduledExecutorService() {
+ return Utils.daemonScheduledThreadPool(
+ concurrentScheduledTaskThreads(), "josdk-scheduled-task");
+ }
+
+ /**
+ * Override to provide a custom {@link ScheduledExecutorService} implementation to change how the
+ * operator's retried and rescheduled reconciliations are triggered. This executor is kept
+ * separate from the one the polling event sources use, see {@link
+ * #getScheduledExecutorService()}, so that a slow poll can't delay a retry.
+ *
+ *
The same notes as for {@link #getScheduledExecutorService()} apply to custom
+ * implementations.
+ *
+ * @return the {@link ScheduledExecutorService} implementation to use to trigger retried and
+ * rescheduled reconciliations
+ * @since 5.6.0
+ */
+ default ScheduledExecutorService getRetryAndRescheduleExecutorService() {
+ return Utils.daemonScheduledThreadPool(
+ concurrentRetryAndRescheduleThreads(), "josdk-retry-reschedule");
+ }
+
/**
* Determines whether the associated Kubernetes client should be closed when the associated {@link
* io.javaoperatorsdk.operator.Operator} is stopped.
@@ -539,4 +621,18 @@ default InformerPool informerPool() {
pool.setConfigurationService(this);
return pool;
}
+
+ /**
+ * Retrieves the {@link DesiredStateAspect}s applied to the desired state of all the Kubernetes
+ * dependent resources managed by the operator. Aspects are applied in the order in which they are
+ * returned, right after the desired state has been computed, and are typically used to add common
+ * metadata (such as a label identifying the operator managing the resource) to all the resources
+ * the operator creates or updates.
+ *
+ * @return the list of aspects to apply to computed desired states, empty by default
+ * @since 5.6.0
+ */
+ default List 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..9ac8668b20 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,9 +16,12 @@
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;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Function;
import org.slf4j.Logger;
@@ -31,6 +34,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"})
@@ -42,11 +46,15 @@ public class ConfigurationServiceOverrider {
private Boolean checkCR;
private Integer concurrentReconciliationThreads;
private Integer concurrentWorkflowExecutorThreads;
+ private Integer concurrentScheduledTaskThreads;
+ private Integer concurrentRetryAndRescheduleThreads;
private Cloner cloner;
private Boolean closeClientOnStop;
private KubernetesClient client;
private ExecutorService executorService;
private ExecutorService workflowExecutorService;
+ private ScheduledExecutorService scheduledExecutorService;
+ private ScheduledExecutorService retryAndRescheduleExecutorService;
private LeaderElectionConfiguration leaderElectionConfiguration;
private String clusterScopedEventNamespace;
private EventRecorder eventRecorder;
@@ -59,6 +67,7 @@ public class ConfigurationServiceOverrider {
private Boolean useSSAToPatchPrimaryResource;
private Boolean cloneSecondaryResourcesWhenGettingFromCache;
private InformerPool informerPool;
+ private List desiredStateAspects;
@SuppressWarnings("rawtypes")
private DependentResourceFactory dependentResourceFactory;
@@ -82,6 +91,34 @@ public ConfigurationServiceOverrider withConcurrentWorkflowExecutorThreads(int t
return this;
}
+ /**
+ * Sets the number of threads used to run the operator's scheduled (i.e. periodic or delayed)
+ * tasks, which are shared by all its polling event sources.
+ *
+ * @param threadNumber the maximum number of concurrent scheduled task threads
+ * @return this {@link ConfigurationServiceOverrider} for chained customization
+ * @see ConfigurationService#concurrentScheduledTaskThreads()
+ * @since 5.6.0
+ */
+ public ConfigurationServiceOverrider withConcurrentScheduledTaskThreads(int threadNumber) {
+ this.concurrentScheduledTaskThreads = threadNumber;
+ return this;
+ }
+
+ /**
+ * Sets the number of threads used to trigger the operator's retried and rescheduled
+ * reconciliations, which are shared by all its controllers.
+ *
+ * @param threadNumber the maximum number of concurrent retry and reschedule threads
+ * @return this {@link ConfigurationServiceOverrider} for chained customization
+ * @see ConfigurationService#concurrentRetryAndRescheduleThreads()
+ * @since 5.6.0
+ */
+ public ConfigurationServiceOverrider withConcurrentRetryAndRescheduleThreads(int threadNumber) {
+ this.concurrentRetryAndRescheduleThreads = threadNumber;
+ return this;
+ }
+
@SuppressWarnings("rawtypes")
public ConfigurationServiceOverrider withDependentResourceFactory(
DependentResourceFactory dependentResourceFactory) {
@@ -115,6 +152,37 @@ public ConfigurationServiceOverrider withWorkflowExecutorService(
return this;
}
+ /**
+ * Replaces the executor used to run the operator's scheduled (i.e. periodic or delayed) tasks,
+ * which are shared by all its polling event sources.
+ *
+ * @param scheduledExecutorService the executor to run scheduled tasks on
+ * @return this {@link ConfigurationServiceOverrider} for chained customization
+ * @see ConfigurationService#getScheduledExecutorService()
+ * @since 5.6.0
+ */
+ public ConfigurationServiceOverrider withScheduledExecutorService(
+ ScheduledExecutorService scheduledExecutorService) {
+ this.scheduledExecutorService = scheduledExecutorService;
+ return this;
+ }
+
+ /**
+ * Replaces the executor used to trigger the operator's retried and rescheduled reconciliations,
+ * which is shared by all its controllers.
+ *
+ * @param retryAndRescheduleExecutorService the executor to trigger retried and rescheduled
+ * reconciliations on
+ * @return this {@link ConfigurationServiceOverrider} for chained customization
+ * @see ConfigurationService#getRetryAndRescheduleExecutorService()
+ * @since 5.6.0
+ */
+ public ConfigurationServiceOverrider withRetryAndRescheduleExecutorService(
+ ScheduledExecutorService retryAndRescheduleExecutorService) {
+ this.retryAndRescheduleExecutorService = retryAndRescheduleExecutorService;
+ return this;
+ }
+
/**
* 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 +297,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
@@ -276,6 +376,28 @@ public int concurrentWorkflowExecutorThreads() {
original.concurrentWorkflowExecutorThreads());
}
+ @Override
+ public int concurrentScheduledTaskThreads() {
+ return Utils.ensureValid(
+ overriddenValueOrDefault(
+ concurrentScheduledTaskThreads,
+ ConfigurationService::concurrentScheduledTaskThreads),
+ "maximum scheduled task threads",
+ 1,
+ original.concurrentScheduledTaskThreads());
+ }
+
+ @Override
+ public int concurrentRetryAndRescheduleThreads() {
+ return Utils.ensureValid(
+ overriddenValueOrDefault(
+ concurrentRetryAndRescheduleThreads,
+ ConfigurationService::concurrentRetryAndRescheduleThreads),
+ "maximum retry and reschedule threads",
+ 1,
+ original.concurrentRetryAndRescheduleThreads());
+ }
+
@Override
public Metrics getMetrics() {
return overriddenValueOrDefault(metrics, ConfigurationService::getMetrics);
@@ -304,6 +426,24 @@ public ExecutorService getWorkflowExecutorService() {
}
}
+ @Override
+ public ScheduledExecutorService getScheduledExecutorService() {
+ if (scheduledExecutorService != null) {
+ return scheduledExecutorService;
+ } else {
+ return super.getScheduledExecutorService();
+ }
+ }
+
+ @Override
+ public ScheduledExecutorService getRetryAndRescheduleExecutorService() {
+ if (retryAndRescheduleExecutorService != null) {
+ return retryAndRescheduleExecutorService;
+ } else {
+ return super.getRetryAndRescheduleExecutorService();
+ }
+ }
+
@Override
public Optional getLeaderElectionConfiguration() {
return leaderElectionConfiguration != null
@@ -383,6 +523,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..437bbfeae9 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
@@ -42,6 +42,7 @@ public class ExecutorServiceManager {
private ExecutorService workflowExecutor;
private ExecutorService cachingExecutorService;
private ScheduledExecutorService scheduledExecutorService;
+ private ScheduledExecutorService retryAndRescheduleExecutorService;
private boolean started;
private ConfigurationService configurationService;
@@ -128,30 +129,54 @@ public ExecutorService cachingExecutorService() {
return cachingExecutorService;
}
+ /**
+ * The executor the operator runs its scheduled (i.e. periodic or delayed) tasks on, shared by its
+ * polling event sources. Note that it is only valid while the manager is started: it is shut down
+ * by {@link #stop(Duration)} and replaced by a fresh one on the next {@link
+ * #start(ConfigurationService)}, so callers should retrieve it when they start rather than hold
+ * on to it.
+ *
+ * @return the executor to run scheduled tasks on
+ */
public ScheduledExecutorService scheduledExecutorService() {
return scheduledExecutorService;
}
+ /**
+ * The executor the operator triggers its retried and rescheduled reconciliations on, kept
+ * separate from {@link #scheduledExecutorService()} so that a slow poll can't delay a retry. The
+ * same lifecycle caveat as for {@link #scheduledExecutorService()} applies.
+ *
+ * @return the executor to trigger retried and rescheduled reconciliations on
+ */
+ public ScheduledExecutorService retryAndRescheduleExecutorService() {
+ return retryAndRescheduleExecutorService;
+ }
+
public synchronized void start(ConfigurationService configurationService) {
if (!started) {
this.configurationService = configurationService; // used to lazy init workflow executor
this.cachingExecutorService = Executors.newCachedThreadPool();
- this.scheduledExecutorService = Executors.newScheduledThreadPool(0);
+ this.scheduledExecutorService = configurationService.getScheduledExecutorService();
+ this.retryAndRescheduleExecutorService =
+ configurationService.getRetryAndRescheduleExecutorService();
this.executor = new InstrumentedExecutorService(configurationService.getExecutorService());
started = true;
}
}
public synchronized void stop(Duration gracefulShutdownTimeout) {
- var parallelExec = Executors.newFixedThreadPool(4);
+ var shutdowns =
+ List.of(
+ shutdown(executor, gracefulShutdownTimeout),
+ shutdown(workflowExecutor, gracefulShutdownTimeout),
+ shutdown(cachingExecutorService, gracefulShutdownTimeout),
+ shutdown(scheduledExecutorService, gracefulShutdownTimeout),
+ shutdown(retryAndRescheduleExecutorService, gracefulShutdownTimeout));
+ var parallelExec = Executors.newFixedThreadPool(shutdowns.size());
try {
log.debug("Closing executor");
- parallelExec.invokeAll(
- List.of(
- shutdown(executor, gracefulShutdownTimeout),
- shutdown(workflowExecutor, gracefulShutdownTimeout),
- shutdown(cachingExecutorService, gracefulShutdownTimeout),
- shutdown(scheduledExecutorService, gracefulShutdownTimeout)));
+ parallelExec.invokeAll(shutdowns);
} catch (InterruptedException e) {
log.debug("Exception closing executor: {}", e.getLocalizedMessage());
Thread.currentThread().interrupt();
diff --git a/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Utils.java b/operator-framework-core/src/main/java/io/javaoperatorsdk/operator/api/config/Utils.java
index 6ad4928c86..95e4fba6e1 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
@@ -26,6 +26,11 @@
import java.util.Date;
import java.util.Optional;
import java.util.Properties;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicLong;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -42,6 +47,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 +59,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 +71,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;
@@ -102,6 +113,49 @@ public static int ensureValid(int value, String description, int minValue, int d
return value;
}
+ /**
+ * Creates a {@link ThreadFactory} producing daemon threads named after the specified prefix.
+ * Daemon threads don't keep the JVM alive if the {@link io.javaoperatorsdk.operator.Operator} is
+ * never stopped, and naming them makes the pool they belong to identifiable in thread dumps.
+ *
+ * @param namePrefix the prefix the created threads are named after
+ * @return a {@link ThreadFactory} creating named daemon threads
+ * @since 5.6.0
+ */
+ public static ThreadFactory daemonThreadFactory(String namePrefix) {
+ final var defaultThreadFactory = Executors.defaultThreadFactory();
+ final var counter = new AtomicLong();
+ return runnable -> {
+ final var thread = defaultThreadFactory.newThread(runnable);
+ thread.setName(namePrefix + "-" + counter.incrementAndGet());
+ thread.setDaemon(true);
+ return thread;
+ };
+ }
+
+ /**
+ * Creates the kind of {@link ScheduledExecutorService} the operator runs its scheduled (i.e.
+ * periodic or delayed) tasks on: one whose threads are daemon threads named after the specified
+ * prefix, and which doesn't let the tasks that were scheduled for later delay its shutdown.
+ *
+ * @param corePoolSize the number of threads to keep in the pool
+ * @param threadNamePrefix the prefix the threads of the pool are named after
+ * @return a {@link ScheduledExecutorService} to run scheduled tasks on
+ * @since 5.6.0
+ */
+ public static ScheduledExecutorService daemonScheduledThreadPool(
+ int corePoolSize, String threadNamePrefix) {
+ final var executor =
+ new ScheduledThreadPoolExecutor(corePoolSize, daemonThreadFactory(threadNamePrefix));
+ // tasks that are scheduled far out (a reconciliation rescheduled in an hour, say) would
+ // otherwise keep the pool from terminating until the graceful shutdown timeout expires
+ executor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+ // cancelled tasks are frequent (every retry that is superseded by a new event cancels one) and
+ // would otherwise be retained until their delay elapses
+ executor.setRemoveOnCancelPolicy(true);
+ return executor;
+ }
+
@SuppressWarnings("unused")
// this is used in the Quarkus extension
public static boolean isValidateCustomResourcesEnvVarSet() {
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