[WIP] feat: schedule polling and retries on managed executors instead of Timers by csviri · Pull Request #3620 · operator-framework/java-operator-sdk · GitHub
Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions docs/content/en/docs/documentation/eventing.md
Original file line number Diff line number Diff line change
Expand Up @@ -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 |
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
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;
Expand Down Expand Up @@ -64,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
Expand Down Expand Up @@ -219,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
*
Expand Down Expand Up @@ -249,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.
*
* <p>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.
*
* <p>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.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ScheduledExecutorService;
import java.util.function.Function;

import org.slf4j.Logger;
Expand All @@ -45,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;
Expand Down Expand Up @@ -86,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) {
Expand Down Expand Up @@ -119,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.
Expand Down Expand Up @@ -312,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);
Expand Down Expand Up @@ -340,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<LeaderElectionConfiguration> getLeaderElectionConfiguration() {
return leaderElectionConfiguration != null
Expand Down
Loading
Loading