This is an automated email from the ASF dual-hosted git repository. sureshanaparti pushed a commit to branch 4.22 in repository https://gitbox.apache.org/repos/asf/cloudstack.git
commit d02adcde9008729c55e648c4bbbd861097aa33bb Merge: 56306ab52cc 2ebcaf626cb Author: Suresh Kumar Anaparti <[email protected]> AuthorDate: Tue Nov 3 17:35:23 2026 +0530 Merge branch '4.20' into 4.22 .../main/java/com/cloud/utils/db/Merovingian2.java | 10 ++ .../cloud/utils/db/Merovingian2AcquireTest.java | 82 ++++++++++++++ .../framework/jobs/impl/AsyncJobManagerImpl.java | 66 +++++++---- .../jobs/impl/AsyncJobManagerImplTest.java | 125 +++++++++++++++++++++ 4 files changed, 262 insertions(+), 21 deletions(-) diff --cc framework/jobs/src/test/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImplTest.java index 08d4c2afd43,364cc7d7a00..7223b8d07f5 --- 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 @@@ -26,7 -21,7 +26,8 @@@ 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.db.SearchCriteria; import com.cloud.utils.fsm.NoTransitionException; import com.cloud.vm.VMInstanceVO; import com.cloud.vm.VirtualMachine; @@@ -37,12 -31,10 +38,14 @@@ import org.apache.cloudstack.api.ApiCom 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.apache.cloudstack.framework.jobs.dao.AsyncJobDao; +import org.junit.After; +import org.junit.Assert; +import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; + import org.mockito.InOrder; import org.mockito.InjectMocks; import org.mockito.Mock; import org.mockito.Mockito; @@@ -72,31 -62,30 +79,54 @@@ public class AsyncJobManagerImplTest NetworkDao networkDao; @Mock NetworkOrchestrationService networkOrchestrationService; + @Mock + AsyncJobDao asyncJobDao; + + private static final long NETWORK_ID = 42L; + private static final long ACCOUNT_ID = 2L; + // job.expire.minutes defaults to one day; these networks were created two days ago + private static final Date OLD_NETWORK_CREATED = new Date(System.currentTimeMillis() - 2 * 24 * 60 * 60 * 1000L); + + private NetworkVO mockImplementingNetworkListed(Date created) { + NetworkVO listed = Mockito.mock(NetworkVO.class); + when(listed.getCreated()).thenReturn(created); + SearchCriteria<NetworkVO> sc = Mockito.mock(SearchCriteria.class); + when(networkDao.createSearchCriteria()).thenReturn(sc); + when(networkDao.search(sc, null)).thenReturn(List.of(listed)); + return listed; + } + + private NetworkVO mockNetworkReadUnderLock(Network.State state) { + NetworkVO locked = Mockito.mock(NetworkVO.class); + when(locked.getState()).thenReturn(state); + when(networkDao.acquireInLockTable(NETWORK_ID, 0)).thenReturn(locked); + return locked; + } + @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(); @@@ -135,108 -124,98 +165,203 @@@ Network.Event.OperationFailed); } + @Test + public void testCleanupOrphanedNetworksSkipsNetworkWhoseLockIsHeld() throws NoTransitionException { + NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED); + when(listed.getId()).thenReturn(NETWORK_ID); + // a VM start implementing the network holds its lock, so a try without waiting fails + when(networkDao.acquireInLockTable(NETWORK_ID, 0)).thenReturn(null); + + asyncJobManager.cleanupOrphanedNetworks(); + + Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any()); + Mockito.verify(networkDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any()); + Mockito.verify(networkDao, Mockito.never()).releaseFromLockTable(Mockito.anyLong()); + Mockito.verifyNoInteractions(asyncJobDao); + } + + @Test + public void testCleanupOrphanedNetworksShutsDownOrphanedNetworkUnderLock() throws NoTransitionException { + NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED); + when(listed.getId()).thenReturn(NETWORK_ID); + NetworkVO locked = mockNetworkReadUnderLock(Network.State.Implementing); + when(locked.getId()).thenReturn(NETWORK_ID); + when(locked.getAccountId()).thenReturn(ACCOUNT_ID); + when(locked.getCreated()).thenReturn(OLD_NETWORK_CREATED); + when(asyncJobDao.findInstancePendingAsyncJobs("Network", ACCOUNT_ID)).thenReturn(Collections.emptyList()); + + asyncJobManager.cleanupOrphanedNetworks(); + + InOrder inOrder = Mockito.inOrder(networkDao, networkOrchestrationService); + inOrder.verify(networkDao).acquireInLockTable(NETWORK_ID, 0); + inOrder.verify(networkOrchestrationService).stateTransitTo(locked, Network.Event.OperationFailed); + inOrder.verify(networkDao).releaseFromLockTable(NETWORK_ID); + Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.eq(listed), Mockito.any()); + } + + @Test + public void testCleanupOrphanedNetworksSkipsNetworkImplementedBeforeItWasLocked() throws NoTransitionException { + NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED); + when(listed.getId()).thenReturn(NETWORK_ID); + // the implementation finished between the search and the lock + mockNetworkReadUnderLock(Network.State.Implemented); + + asyncJobManager.cleanupOrphanedNetworks(); + + Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any()); + Mockito.verify(networkDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any()); + Mockito.verify(networkDao).releaseFromLockTable(NETWORK_ID); + } + + @Test + public void testCleanupOrphanedNetworksSkipsNetworkWithPendingNetworkJob() throws NoTransitionException { + NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED); + when(listed.getId()).thenReturn(NETWORK_ID); + NetworkVO locked = mockNetworkReadUnderLock(Network.State.Implementing); + when(locked.getId()).thenReturn(NETWORK_ID); + when(locked.getAccountId()).thenReturn(ACCOUNT_ID); + AsyncJobVO job = new AsyncJobVO(); + job.setInstanceId(NETWORK_ID); + when(asyncJobDao.findInstancePendingAsyncJobs("Network", ACCOUNT_ID)).thenReturn(List.of(job)); + + asyncJobManager.cleanupOrphanedNetworks(); + + Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any()); + Mockito.verify(networkDao).releaseFromLockTable(NETWORK_ID); + } + + @Test + public void testCleanupOrphanedNetworksSkipsYoungNetwork() throws NoTransitionException { + mockImplementingNetworkListed(new Date()); + + asyncJobManager.cleanupOrphanedNetworks(); + + Mockito.verify(networkDao, Mockito.never()).acquireInLockTable(Mockito.anyLong(), Mockito.anyInt()); + Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any()); + Mockito.verifyNoInteractions(asyncJobDao); + } + + @Test + public void testCleanupOrphanedNetworksReleasesLockWhenForcingShutdown() throws NoTransitionException { + NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED); + when(listed.getId()).thenReturn(NETWORK_ID); + NetworkVO locked = mockNetworkReadUnderLock(Network.State.Implementing); + when(locked.getId()).thenReturn(NETWORK_ID); + when(locked.getAccountId()).thenReturn(ACCOUNT_ID); + when(locked.getCreated()).thenReturn(OLD_NETWORK_CREATED); + when(asyncJobDao.findInstancePendingAsyncJobs("Network", ACCOUNT_ID)).thenReturn(Collections.emptyList()); + when(networkOrchestrationService.stateTransitTo(locked, Network.Event.OperationFailed)).thenThrow(new NoTransitionException("no transition")); + + asyncJobManager.cleanupOrphanedNetworks(); + + InOrder inOrder = Mockito.inOrder(locked, networkDao); + inOrder.verify(locked).setState(Network.State.Shutdown); + inOrder.verify(networkDao).update(NETWORK_ID, locked); + inOrder.verify(networkDao).releaseFromLockTable(NETWORK_ID); + } ++ + @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); + } }
