Do not shut down networks that are being implemented in the orphaned-network cleanup

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)) {

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

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<NetworkVO> sc = networkDao.createSearchCriteria();
             sc.addAnd("state", SearchCriteria.Op.EQ, Network.State.Implementing);
@@ -1313,30 +1313,54 @@ private void cleanupOrphanedNetworks() {
                     continue;
                 }
 
-                List<AsyncJobVO> 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<AsyncJobVO> 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);
         }
     }
 

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