This is an automated email from the ASF dual-hosted git repository.
nvazquez pushed a commit to branch 4.22
in repository https://gitbox.apache.org/repos/asf/cloudstack.git
The following commit(s) were added to refs/heads/4.22 by this push:
new 2974af84f22 Make worker and API thread pool size configurable (#13565)
2974af84f22 is described below
commit 2974af84f22a8def8339380628d8653a4f016d3f
Author: Daman Arora <[email protected]>
AuthorDate: Tue Oct 6 15:21:42 2026 -0400
Make worker and API thread pool size configurable (#13565)
* Make worker and API thread pool size configurable
The AsyncJobManager's API and Worker executor thread pool sizes are
currently derived solely from db.cloud.maxActive (maxActive/2 and
maxActive*2/3 respectively). This coupling doesn't account for
environments with predominantly I/O-bound workers, where threads are
held waiting on external responses for extended periods and higher
pool sizes may be needed independently of the DB connection count.
This adds two global config keys:
- api.job.pool.size (default: 50)
- work.job.pool.size (default: 50)
The actual pool size is Math.max(configured value, db-derived
default), preserving existing behavior by default while allowing
operators to increase pool sizes when workloads require it. Pool
sizes and their derivation are logged at startup for observability.
* Replace synthetic pool size tests with real configure() executor
assertions
The previous unit tests for the api.job.pool.size/work.job.pool.size
config keys only re-ran Math.max() on local hardcoded values, or
checked ConfigKey plumbing directly, without ever calling
AsyncJobManagerImpl.configure() itself. That meant a regression in the
actual pool-sizing logic wouldn't have been caught.
These tests now invoke configure() directly and assert on the real
core pool size of the created _apiJobExecutor/_workerJobExecutor via
reflection, with @Before/@After hooks to stub db.properties and reset
config/executor state between tests.
---------
Co-authored-by: Sachin R <[email protected]>
Co-authored-by: mprokopchuk <[email protected]>
---
.../framework/jobs/impl/AsyncJobManagerImpl.java | 19 ++-
.../jobs/impl/AsyncJobManagerImplTest.java | 146 +++++++++++++++++++++
2 files changed, 160 insertions(+), 5 deletions(-)
diff --git
a/framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
b/framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
index 1cb1cb4e309..1572adc5008 100644
---
a/framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
+++
b/framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java
@@ -125,6 +125,10 @@ public class AsyncJobManagerImpl extends ManagerBase
implements AsyncJobManager,
Integer.class, "vm.job.lock.timeout", "1800",
"Time in seconds to wait in acquiring lock to submit a vm worker
job", false);
private static final ConfigKey<Boolean> HidePassword = new
ConfigKey<Boolean>("Advanced", Boolean.class, "log.hide.password", "true", "If
set to true, the password is hidden", true, ConfigKey.Scope.Global);
+ public static final ConfigKey<Integer> ApiJobPoolSize = new
ConfigKey<>("Advanced", Integer.class, "api.job.pool.size", "50",
+ "Minimum size of the API job executor thread pool. Actual size is the
max of this value and db.cloud.maxActive / 2.", false, ConfigKey.Scope.Global);
+ public static final ConfigKey<Integer> WorkJobPoolSize = new
ConfigKey<>("Advanced", Integer.class, "work.job.pool.size", "50",
+ "Minimum size of the Worker job executor thread pool. Actual size is
the max of this value and db.cloud.maxActive * 2 / 3.", false,
ConfigKey.Scope.Global);
private static final int ACQUIRE_GLOBAL_LOCK_TIMEOUT_FOR_COOPERATION = 3;
// 3 seconds
@@ -198,7 +202,7 @@ public class AsyncJobManagerImpl extends ManagerBase
implements AsyncJobManager,
@Override
public ConfigKey<?>[] getConfigKeys() {
- return new ConfigKey<?>[] {JobExpireMinutes,
JobCancelThresholdMinutes, VmJobLockTimeout, HidePassword};
+ return new ConfigKey<?>[] {JobExpireMinutes,
JobCancelThresholdMinutes, VmJobLockTimeout, HidePassword, ApiJobPoolSize,
WorkJobPoolSize};
}
@Override
@@ -1100,13 +1104,18 @@ public class AsyncJobManagerImpl extends ManagerBase
implements AsyncJobManager,
final Properties dbProps = DbProperties.getDbProperties();
final int cloudMaxActive =
Integer.parseInt(dbProps.getProperty("db.cloud.maxActive"));
- int apiPoolSize = cloudMaxActive / 2;
- int workPoolSize = (cloudMaxActive * 2) / 3;
+ int defaultApiPoolSize = cloudMaxActive / 2;
+ int defaultWorkPoolSize = (cloudMaxActive * 2) / 3;
- logger.info("Start AsyncJobManager API executor thread pool in
size " + apiPoolSize);
+ int apiPoolSize = Math.max(ApiJobPoolSize.value(),
defaultApiPoolSize);
+ int workPoolSize = Math.max(WorkJobPoolSize.value(),
defaultWorkPoolSize);
+
+ logger.info("Start AsyncJobManager API executor thread pool in
size " + apiPoolSize +
+ " (configured=" + ApiJobPoolSize.value() + ", db.derived=" +
defaultApiPoolSize + ", db.cloud.maxActive=" + cloudMaxActive + ")");
_apiJobExecutor = Executors.newFixedThreadPool(apiPoolSize, new
NamedThreadFactory(AsyncJobManager.API_JOB_POOL_THREAD_PREFIX));
- logger.info("Start AsyncJobManager Work executor thread pool in
size " + workPoolSize);
+ logger.info("Start AsyncJobManager Work executor thread pool in
size " + workPoolSize +
+ " (configured=" + WorkJobPoolSize.value() + ", db.derived=" +
defaultWorkPoolSize + ", db.cloud.maxActive=" + cloudMaxActive + ")");
_workerJobExecutor = Executors.newFixedThreadPool(workPoolSize,
new NamedThreadFactory(AsyncJobManager.WORK_JOB_POOL_THREAD_PREFIX));
} catch (final Exception e) {
throw new ConfigurationException("Unable to load db.properties to
configure AsyncJobManagerImpl");
diff --git
a/framework/jobs/src/test/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImplTest.java
b/framework/jobs/src/test/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImplTest.java
index 0be5dbc01cb..08d4c2afd43 100644
---
a/framework/jobs/src/test/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImplTest.java
+++
b/framework/jobs/src/test/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImplTest.java
@@ -17,19 +17,30 @@
package org.apache.cloudstack.framework.jobs.impl;
+import java.lang.reflect.Field;
+import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ThreadPoolExecutor;
+
import com.cloud.network.Network;
import com.cloud.network.dao.NetworkDao;
import com.cloud.network.dao.NetworkVO;
import com.cloud.storage.Volume;
+import com.cloud.utils.db.DbProperties;
import com.cloud.utils.fsm.NoTransitionException;
import com.cloud.vm.VMInstanceVO;
import com.cloud.vm.VirtualMachine;
import com.cloud.vm.VirtualMachineManager;
import com.cloud.vm.dao.VMInstanceDao;
+
import org.apache.cloudstack.api.ApiCommandResourceType;
import
org.apache.cloudstack.engine.orchestration.service.NetworkOrchestrationService;
import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
+import org.apache.cloudstack.framework.config.ConfigKey;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
@@ -42,6 +53,12 @@ import static org.mockito.Mockito.when;
@RunWith(MockitoJUnitRunner.class)
public class AsyncJobManagerImplTest {
+
+ // db.cloud.maxActive value preloaded into DbProperties for every test in
this class.
+ // Chosen so the db-derived defaults (4000/2 = 2000 for api, 4000*2/3 =
2666 for worker)
+ // are non-trivial and easy to reason about vs. the configured minimums
under test.
+ private static final int DB_CLOUD_MAX_ACTIVE = 4000;
+
@Spy
@InjectMocks
AsyncJobManagerImpl asyncJobManager;
@@ -56,6 +73,30 @@ public class AsyncJobManagerImplTest {
@Mock
NetworkOrchestrationService networkOrchestrationService;
+ @Before
+ public void preloadDbProperties() throws Exception {
+ Properties props = new Properties();
+ props.setProperty("db.cloud.maxActive",
String.valueOf(DB_CLOUD_MAX_ACTIVE));
+ // Bypass DbProperties' internal "loaded" guard so configure() reads
our
+ // properties instead of trying to load a real db.properties file.
+ setDbPropertiesStatics(props, true);
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ try {
+ shutdownIfPresent("_apiJobExecutor");
+ shutdownIfPresent("_workerJobExecutor");
+ } finally {
+ // Reset static state even if executor shutdown throws, so we do
not
+ // pollute later test classes with our stubbed ConfigKey values or
+ // preloaded DbProperties.
+ resetConfigValue(AsyncJobManagerImpl.ApiJobPoolSize);
+ resetConfigValue(AsyncJobManagerImpl.WorkJobPoolSize);
+ setDbPropertiesStatics(new Properties(), false);
+ }
+ }
+
@Test
public void testCleanupVolumeResource() {
AsyncJobVO job = new AsyncJobVO();
@@ -93,4 +134,109 @@ public class AsyncJobManagerImplTest {
Mockito.verify(networkOrchestrationService,
Mockito.times(1)).stateTransitTo(networkVO,
Network.Event.OperationFailed);
}
+
+ @Test
+ public void configureApiPoolUsesConfiguredValueWhenLargerThanDbDerived()
throws Exception {
+ overrideConfigValue(AsyncJobManagerImpl.ApiJobPoolSize, 3000);
+ overrideConfigValue(AsyncJobManagerImpl.WorkJobPoolSize, 50);
+ invokeConfigureAndSwallowPostSetupNpe();
+ // apiPoolSize = max(3000, 4000/2) = 3000 (configured wins)
+ Assert.assertEquals(3000, apiPoolCoreSize());
+ }
+
+ @Test
+ public void configureApiPoolUsesDbDerivedWhenLargerThanConfigured() throws
Exception {
+ overrideConfigValue(AsyncJobManagerImpl.ApiJobPoolSize, 50);
+ overrideConfigValue(AsyncJobManagerImpl.WorkJobPoolSize, 50);
+ invokeConfigureAndSwallowPostSetupNpe();
+ // apiPoolSize = max(50, 4000/2) = 2000 (db-derived wins)
+ Assert.assertEquals(DB_CLOUD_MAX_ACTIVE / 2, apiPoolCoreSize());
+ }
+
+ @Test
+ public void configureWorkerPoolUsesTwoThirdsFormula() throws Exception {
+ overrideConfigValue(AsyncJobManagerImpl.ApiJobPoolSize, 50);
+ overrideConfigValue(AsyncJobManagerImpl.WorkJobPoolSize, 50);
+ invokeConfigureAndSwallowPostSetupNpe();
+ // workPoolSize = max(50, 4000*2/3) = 2666 (db-derived wins)
+ Assert.assertEquals((DB_CLOUD_MAX_ACTIVE * 2) / 3,
workerPoolCoreSize());
+ }
+
+ @Test
+ public void
configureWorkerPoolUsesConfiguredValueWhenLargerThanDbDerived() throws
Exception {
+ overrideConfigValue(AsyncJobManagerImpl.ApiJobPoolSize, 50);
+ overrideConfigValue(AsyncJobManagerImpl.WorkJobPoolSize, 5000);
+ invokeConfigureAndSwallowPostSetupNpe();
+ // workPoolSize = max(5000, 4000*2/3) = 5000 (configured wins)
+ Assert.assertEquals(5000, workerPoolCoreSize());
+ }
+
+ /**
+ * Invoke configure() and swallow the NPE it throws when wiring
SearchBuilders
+ * on the un-mocked DAOs after the pool-sizing block. The executors are
created
+ * inside configure()'s try/catch (before the DAO wiring), so pool sizes
can
+ * still be inspected after the failure.
+ */
+ private void invokeConfigureAndSwallowPostSetupNpe() {
+ try {
+ asyncJobManager.configure(null, null);
+ } catch (Exception ignored) {
+ // expected: DAO wiring after pool-sizing NPEs because we
intentionally
+ // did not mock the DAOs (they are not part of what we are
testing).
+ }
+ }
+
+ private int apiPoolCoreSize() throws Exception {
+ Object executor = readField("_apiJobExecutor");
+ Assert.assertNotNull("configure() did not create _apiJobExecutor - has
DAO wiring been moved before pool sizing?", executor);
+ return ((ThreadPoolExecutor) executor).getCorePoolSize();
+ }
+
+ private int workerPoolCoreSize() throws Exception {
+ Object executor = readField("_workerJobExecutor");
+ Assert.assertNotNull("configure() did not create _workerJobExecutor -
has DAO wiring been moved before pool sizing?", executor);
+ return ((ThreadPoolExecutor) executor).getCorePoolSize();
+ }
+
+ private Object readField(String name) throws Exception {
+ Field f = AsyncJobManagerImpl.class.getDeclaredField(name);
+ f.setAccessible(true);
+ return f.get(asyncJobManager);
+ }
+
+ private void shutdownIfPresent(String executorFieldName) {
+ try {
+ Object executor = readField(executorFieldName);
+ if (executor instanceof ExecutorService) {
+ ((ExecutorService) executor).shutdownNow();
+ }
+ } catch (Exception ignored) {
+ // never populated by this test - nothing to clean up
+ }
+ }
+
+ private void overrideConfigValue(final ConfigKey<?> configKey, final
Object value) {
+ try {
+ Field f = ConfigKey.class.getDeclaredField("_value");
+ f.setAccessible(true);
+ f.set(configKey, value);
+ } catch (IllegalAccessException | NoSuchFieldException e) {
+ Assert.fail(e.getMessage());
+ }
+ }
+
+ private void resetConfigValue(final ConfigKey<?> configKey) throws
Exception {
+ Field f = ConfigKey.class.getDeclaredField("_value");
+ f.setAccessible(true);
+ f.set(configKey, null);
+ }
+
+ private void setDbPropertiesStatics(Properties props, boolean loaded)
throws Exception {
+ Field propsField = DbProperties.class.getDeclaredField("properties");
+ Field loadedField = DbProperties.class.getDeclaredField("loaded");
+ propsField.setAccessible(true);
+ loadedField.setAccessible(true);
+ propsField.set(null, props);
+ loadedField.set(null, loaded);
+ }
}
