Make worker and API thread pool size configurable

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 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<String, Object> 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");

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);
+    }
 }