From 2903a0413d884e4045be8c434276efcd9da31ca2 Mon Sep 17 00:00:00 2001 From: Andrija Panic Date: Thu, 29 Oct 2026 11:13:20 +0100 Subject: [PATCH] Do not shut down networks that are being implemented in the orphaned-network cleanup The cleanup added in #12409 moves a network in Implementing to Shutdown when it is older than job.expire.minutes and no pending Network job refers to it. A VM start implements the network inside the VM's own job, so a cleanup run during the implementation shut the network down under the implementing thread, and the start failed with "Network is not yet fully implemented". The implementing thread holds the network's lock for the whole time the network is in Implementing. Try that lock once, without waiting, skip the network if it is held, and re-check the state on the row read under the lock before changing it. Merovingian2.acquire never tried the lock for a timeout of 0 (the wait loop did not run) and logged a timeout WARN. A timeout of 0 or less now makes exactly one attempt and returns without waiting; positive timeouts are unchanged. Fixes #14522 --- .../java/com/cloud/utils/db/Merovingian2.java | 10 ++ .../utils/db/Merovingian2AcquireTest.java | 82 ++++++++++++ .../jobs/impl/AsyncJobManagerImpl.java | 66 ++++++--- .../jobs/impl/AsyncJobManagerImplTest.java | 125 ++++++++++++++++++ 4 files changed, 262 insertions(+), 21 deletions(-) create mode 100644 framework/db/src/test/java/com/cloud/utils/db/Merovingian2AcquireTest.java 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); + } }