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..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 @@ -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 @@ @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 void testCleanupNetworkResource() throws NoTransitionException { 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); + } }