From 80017fdc65e3a4fa35e9fc46c7d6ff2b3e5b730a Mon Sep 17 00:00:00 2001 From: Sachin R <32716246+sachindoddaguni@users.noreply.github.com> Date: Fri, 8 May 2026 16:41:46 -0700 Subject: [PATCH 1/2] 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. --- .../jobs/impl/AsyncJobManagerImpl.java | 19 ++++++-- .../jobs/impl/AsyncJobManagerImplTest.java | 48 +++++++++++++++++++ 2 files changed, 62 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 9d71c6d07b49..04e004296d63 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 HidePassword = new ConfigKey("Advanced", Boolean.class, "log.hide.password", "true", "If set to true, the password is hidden", true, ConfigKey.Scope.Global); + public static final ConfigKey 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 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 String getConfigComponentName() { @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 boolean configure(String name, Map params) throws Configu 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 0be5dbc01cbd..75f44c4d36ff 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,6 +17,8 @@ package org.apache.cloudstack.framework.jobs.impl; +import java.lang.reflect.Field; + import com.cloud.network.Network; import com.cloud.network.dao.NetworkDao; import com.cloud.network.dao.NetworkVO; @@ -26,10 +28,13 @@ 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.Assert; import org.junit.Test; import org.junit.runner.RunWith; import org.mockito.InjectMocks; @@ -93,4 +98,47 @@ public void testCleanupNetworkResource() throws NoTransitionException { Mockito.verify(networkOrchestrationService, Mockito.times(1)).stateTransitTo(networkVO, Network.Event.OperationFailed); } + + @Test + public void testPoolSizeUsesConfiguredValueWhenLarger() { + int configuredSize = 3000; + int dbDerivedSize = 2000; + int result = Math.max(configuredSize, dbDerivedSize); + Assert.assertEquals(3000, result); + } + + @Test + public void testPoolSizeUsesDbDerivedValueWhenLarger() { + int configuredSize = 50; + int dbDerivedSize = 2000; + int result = Math.max(configuredSize, dbDerivedSize); + Assert.assertEquals(2000, result); + } + + @Test + public void testPoolSizeDefaultConfigKeyValues() { + overrideConfigValue(AsyncJobManagerImpl.ApiJobPoolSize, 50); + overrideConfigValue(AsyncJobManagerImpl.WorkJobPoolSize, 50); + Assert.assertEquals(Integer.valueOf(50), AsyncJobManagerImpl.ApiJobPoolSize.value()); + Assert.assertEquals(Integer.valueOf(50), AsyncJobManagerImpl.WorkJobPoolSize.value()); + } + + @Test + public void testPoolSizeConfiguredOverridesDbDerived() { + overrideConfigValue(AsyncJobManagerImpl.ApiJobPoolSize, 5000); + int dbMaxActive = 4000; + int dbDerived = dbMaxActive / 2; + int actual = Math.max(AsyncJobManagerImpl.ApiJobPoolSize.value(), dbDerived); + Assert.assertEquals(5000, actual); + } + + 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()); + } + } } From 5d500b8aafc96ed9d996959985848adcae26693f Mon Sep 17 00:00:00 2001 From: mprokopchuk Date: Mon, 6 Jul 2026 16:06:11 -0700 Subject: [PATCH 2/2] 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. --- .../jobs/impl/AsyncJobManagerImplTest.java | 138 +++++++++++++++--- 1 file changed, 118 insertions(+), 20 deletions(-) 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 75f44c4d36ff..08d4c2afd431 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 @@ -18,11 +18,15 @@ 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; @@ -34,7 +38,9 @@ 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; @@ -47,6 +53,12 @@ @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; @@ -61,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(); @@ -100,39 +136,86 @@ public void testCleanupNetworkResource() throws NoTransitionException { } @Test - public void testPoolSizeUsesConfiguredValueWhenLarger() { - int configuredSize = 3000; - int dbDerivedSize = 2000; - int result = Math.max(configuredSize, dbDerivedSize); - Assert.assertEquals(3000, result); + 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 testPoolSizeUsesDbDerivedValueWhenLarger() { - int configuredSize = 50; - int dbDerivedSize = 2000; - int result = Math.max(configuredSize, dbDerivedSize); - Assert.assertEquals(2000, result); + 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 testPoolSizeDefaultConfigKeyValues() { + public void configureWorkerPoolUsesTwoThirdsFormula() throws Exception { overrideConfigValue(AsyncJobManagerImpl.ApiJobPoolSize, 50); overrideConfigValue(AsyncJobManagerImpl.WorkJobPoolSize, 50); - Assert.assertEquals(Integer.valueOf(50), AsyncJobManagerImpl.ApiJobPoolSize.value()); - Assert.assertEquals(Integer.valueOf(50), AsyncJobManagerImpl.WorkJobPoolSize.value()); + invokeConfigureAndSwallowPostSetupNpe(); + // workPoolSize = max(50, 4000*2/3) = 2666 (db-derived wins) + Assert.assertEquals((DB_CLOUD_MAX_ACTIVE * 2) / 3, workerPoolCoreSize()); } @Test - public void testPoolSizeConfiguredOverridesDbDerived() { - overrideConfigValue(AsyncJobManagerImpl.ApiJobPoolSize, 5000); - int dbMaxActive = 4000; - int dbDerived = dbMaxActive / 2; - int actual = Math.max(AsyncJobManagerImpl.ApiJobPoolSize.value(), dbDerived); - Assert.assertEquals(5000, actual); + 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 overrideConfigValue(final ConfigKey configKey, final Object value) { + 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); @@ -141,4 +224,19 @@ private void overrideConfigValue(final ConfigKey configKey, final Object value) 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); + } }