diff --git a/framework/db/src/main/java/com/cloud/utils/db/Merovingian2.java b/framework/db/src/main/java/com/cloud/utils/db/Merovingian2.java index a9a1ce4272eb..3ef2456f6ac3 100644 --- a/framework/db/src/main/java/com/cloud/utils/db/Merovingian2.java +++ b/framework/db/src/main/java/com/cloud/utils/db/Merovingian2.java @@ -127,6 +127,16 @@ public boolean acquire(String key, int timeInSeconds) { if (LOGGER.isTraceEnabled()) { LOGGER.trace("Acquiring lck-" + key + " with wait time of " + timeInSeconds); } + + if (timeInSeconds <= 0) { + // no wait: try exactly once (the loop below would not try at all) + int count = owns(key); + if (count >= 1) { + return increment(key, threadName, threadId); + } + return count == 0 && doAcquire(key, threadName, threadId); + } + long startTime = InaccurateClock.getTime(); while ((InaccurateClock.getTime() - startTime) < (timeInSeconds * 1000l)) { diff --git a/framework/db/src/test/java/com/cloud/utils/db/Merovingian2AcquireTest.java b/framework/db/src/test/java/com/cloud/utils/db/Merovingian2AcquireTest.java new file mode 100644 index 000000000000..44f87cde9023 --- /dev/null +++ b/framework/db/src/test/java/com/cloud/utils/db/Merovingian2AcquireTest.java @@ -0,0 +1,82 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. +package com.cloud.utils.db; + +import org.junit.Assert; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mockito; +import org.mockito.junit.MockitoJUnitRunner; + +@RunWith(MockitoJUnitRunner.class) +public class Merovingian2AcquireTest { + private static final String KEY = "networks42"; + + // the lock table calls are stubbed, so no database is needed + private final Merovingian2 lockController = Mockito.mock(Merovingian2.class, Mockito.CALLS_REAL_METHODS); + + @Test + public void acquireWithoutWaitTakesFreeLock() { + Mockito.doReturn(0).when(lockController).owns(KEY); + Mockito.doReturn(true).when(lockController).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt()); + + Assert.assertTrue(lockController.acquire(KEY, 0)); + Mockito.verify(lockController).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt()); + } + + @Test + public void acquireWithoutWaitTriesOnceWhenLockIsHeldElsewhere() { + Mockito.doReturn(-1).when(lockController).owns(KEY); + + Assert.assertFalse(lockController.acquire(KEY, 0)); + Mockito.verify(lockController, Mockito.times(1)).owns(KEY); + Mockito.verify(lockController, Mockito.never()).doAcquire(Mockito.anyString(), Mockito.anyString(), Mockito.anyInt()); + } + + @Test + public void acquireWithoutWaitReentersLockHeldByThisThread() { + Mockito.doReturn(1).when(lockController).owns(KEY); + Mockito.doReturn(true).when(lockController).increment(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt()); + + Assert.assertTrue(lockController.acquire(KEY, 0)); + Mockito.verify(lockController, Mockito.never()).doAcquire(Mockito.anyString(), Mockito.anyString(), Mockito.anyInt()); + } + + @Test + public void acquireWithoutWaitFailsWhenInsertLosesRace() { + Mockito.doReturn(0).when(lockController).owns(KEY); + Mockito.doReturn(false).when(lockController).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt()); + + Assert.assertFalse(lockController.acquire(KEY, 0)); + Mockito.verify(lockController, Mockito.times(1)).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt()); + } + + @Test + public void acquireWithTimeoutStillWaitsForHeldLock() { + // held on the first try, free on the second; the timeout is long enough for the 5 s retry interval + Mockito.doReturn(-1).doReturn(0).when(lockController).owns(KEY); + Mockito.doReturn(true).when(lockController).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt()); + + long start = System.currentTimeMillis(); + Assert.assertTrue(lockController.acquire(KEY, 30)); + long waited = System.currentTimeMillis() - start; + + Assert.assertTrue("waited only " + waited + " ms", waited >= 4900); + Mockito.verify(lockController, Mockito.times(2)).owns(KEY); + Mockito.verify(lockController, Mockito.times(1)).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt()); + } +} 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 f9fdc0751323..9f21971ab8d6 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 @@ -1288,7 +1288,7 @@ private void cleanupFailedSnapshotsCreatedWithDefaultStrategy(final long msid) { * Cleanup networks that are stuck in Implementing state without associated async jobs. * This only processes networks that have been stuck for longer than the job expiration threshold. */ - private void cleanupOrphanedNetworks() { + protected void cleanupOrphanedNetworks() { try { SearchCriteria sc = networkDao.createSearchCriteria(); sc.addAnd("state", SearchCriteria.Op.EQ, Network.State.Implementing); @@ -1313,30 +1313,54 @@ private void cleanupOrphanedNetworks() { continue; } - List jobs = _jobDao.findInstancePendingAsyncJobs("Network", network.getAccountId()); - boolean hasActiveJob = false; - for (AsyncJobVO job : jobs) { - if (job.getInstanceId() != null && job.getInstanceId().equals(network.getId())) { - hasActiveJob = true; - break; - } - } + cleanupNetworkIfOrphaned(network.getId(), expireMinutes); + } + } catch (Exception e) { + logger.error("Error while cleaning up orphaned networks", e); + } + } - if (hasActiveJob) { - logger.debug("Network {} in Implementing state has active async job, skipping cleanup", network.getId()); - continue; - } + /** + * The thread implementing a network holds the network's lock for the whole implementation, and that work can run + * inside a job of another type (a VM start, for example), so a network in Implementing is orphaned only if its lock + * is free. The lock is tried once without waiting, so the GC task does not block behind a running implementation, + * and the network is re-read under the lock before it is changed. + */ + private void cleanupNetworkIfOrphaned(long networkId, long expireMinutes) { + NetworkVO network = networkDao.acquireInLockTable(networkId, 0); + if (network == null) { + logger.debug("Network {} in Implementing state is locked (being implemented) or was removed, skipping cleanup", networkId); + return; + } - logger.warn("Found orphaned network {} in Implementing state without async job. " + - "Network created: {}, age: {} minutes, expiration threshold: {} minutes. Transitioning to Shutdown state.", - network.getId(), network.getCreated(), - (System.currentTimeMillis() - network.getCreated().getTime()) / 60000, - expireMinutes); - updateNetworkState(network); + try { + if (!Network.State.Implementing.equals(network.getState())) { + logger.debug("Network {} is no longer in Implementing state (now {}), skipping cleanup", networkId, network.getState()); + return; + } + List jobs = _jobDao.findInstancePendingAsyncJobs("Network", network.getAccountId()); + boolean hasActiveJob = false; + for (AsyncJobVO job : jobs) { + if (job.getInstanceId() != null && job.getInstanceId().equals(network.getId())) { + hasActiveJob = true; + break; + } } - } catch (Exception e) { - logger.error("Error while cleaning up orphaned networks", e); + + if (hasActiveJob) { + logger.debug("Network {} in Implementing state has active async job, skipping cleanup", network.getId()); + return; + } + + logger.warn("Found orphaned network {} in Implementing state without async job. " + + "Network created: {}, age: {} minutes, expiration threshold: {} minutes. Transitioning to Shutdown state.", + network.getId(), network.getCreated(), + (System.currentTimeMillis() - network.getCreated().getTime()) / 60000, + expireMinutes); + updateNetworkState(network); + } finally { + networkDao.releaseFromLockTable(networkId); } } 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..364cc7d7a004 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 @@ -21,6 +21,7 @@ import com.cloud.network.dao.NetworkDao; import com.cloud.network.dao.NetworkVO; import com.cloud.storage.Volume; +import com.cloud.utils.db.SearchCriteria; import com.cloud.utils.fsm.NoTransitionException; import com.cloud.vm.VMInstanceVO; import com.cloud.vm.VirtualMachine; @@ -30,14 +31,20 @@ 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.jobs.dao.AsyncJobDao; 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; import org.mockito.Spy; import org.mockito.junit.MockitoJUnitRunner; +import java.util.Collections; +import java.util.Date; +import java.util.List; + import static org.mockito.Mockito.when; @RunWith(MockitoJUnitRunner.class) @@ -55,6 +62,29 @@ 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 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; + } @Test public void testCleanupVolumeResource() { @@ -93,4 +123,99 @@ public void testCleanupNetworkResource() throws NoTransitionException { Mockito.verify(networkOrchestrationService, Mockito.times(1)).stateTransitTo(networkVO, 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); + } }