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.

Author: sachindoddaguni

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