Skip to content

Commit 2ebcaf6

Browse files
andrijapanicsbandrijapanicsb
authored
Do not shut down networks that are being implemented in the orphaned-network cleanup (#14527)
1 parent 454bba0 commit 2ebcaf6

4 files changed

Lines changed: 262 additions & 21 deletions

File tree

‎framework/db/src/main/java/com/cloud/utils/db/Merovingian2.java‎

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -127,6 +127,16 @@ public boolean acquire(String key, int timeInSeconds) {
127127
if (LOGGER.isTraceEnabled()) {
128128
LOGGER.trace("Acquiring lck-" + key + " with wait time of " + timeInSeconds);
129129
}
130+
131+
if (timeInSeconds <= 0) {
132+
// no wait: try exactly once (the loop below would not try at all)
133+
int count = owns(key);
134+
if (count >= 1) {
135+
return increment(key, threadName, threadId);
136+
}
137+
return count == 0 && doAcquire(key, threadName, threadId);
138+
}
139+
130140
long startTime = InaccurateClock.getTime();
131141

132142
while ((InaccurateClock.getTime() - startTime) < (timeInSeconds * 1000l)) {

‎framework/db/src/test/java/com/cloud/utils/db/Merovingian2AcquireTest.java‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
package com.cloud.utils.db;
18+
19+
import org.junit.Assert;
20+
import org.junit.Test;
21+
import org.junit.runner.RunWith;
22+
import org.mockito.Mockito;
23+
import org.mockito.junit.MockitoJUnitRunner;
24+
25+
@RunWith(MockitoJUnitRunner.class)
26+
public class Merovingian2AcquireTest {
27+
private static final String KEY = "networks42";
28+
29+
// the lock table calls are stubbed, so no database is needed
30+
private final Merovingian2 lockController = Mockito.mock(Merovingian2.class, Mockito.CALLS_REAL_METHODS);
31+
32+
@Test
33+
public void acquireWithoutWaitTakesFreeLock() {
34+
Mockito.doReturn(0).when(lockController).owns(KEY);
35+
Mockito.doReturn(true).when(lockController).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt());
36+
37+
Assert.assertTrue(lockController.acquire(KEY, 0));
38+
Mockito.verify(lockController).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt());
39+
}
40+
41+
@Test
42+
public void acquireWithoutWaitTriesOnceWhenLockIsHeldElsewhere() {
43+
Mockito.doReturn(-1).when(lockController).owns(KEY);
44+
45+
Assert.assertFalse(lockController.acquire(KEY, 0));
46+
Mockito.verify(lockController, Mockito.times(1)).owns(KEY);
47+
Mockito.verify(lockController, Mockito.never()).doAcquire(Mockito.anyString(), Mockito.anyString(), Mockito.anyInt());
48+
}
49+
50+
@Test
51+
public void acquireWithoutWaitReentersLockHeldByThisThread() {
52+
Mockito.doReturn(1).when(lockController).owns(KEY);
53+
Mockito.doReturn(true).when(lockController).increment(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt());
54+
55+
Assert.assertTrue(lockController.acquire(KEY, 0));
56+
Mockito.verify(lockController, Mockito.never()).doAcquire(Mockito.anyString(), Mockito.anyString(), Mockito.anyInt());
57+
}
58+
59+
@Test
60+
public void acquireWithoutWaitFailsWhenInsertLosesRace() {
61+
Mockito.doReturn(0).when(lockController).owns(KEY);
62+
Mockito.doReturn(false).when(lockController).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt());
63+
64+
Assert.assertFalse(lockController.acquire(KEY, 0));
65+
Mockito.verify(lockController, Mockito.times(1)).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt());
66+
}
67+
68+
@Test
69+
public void acquireWithTimeoutStillWaitsForHeldLock() {
70+
// held on the first try, free on the second; the timeout is long enough for the 5 s retry interval
71+
Mockito.doReturn(-1).doReturn(0).when(lockController).owns(KEY);
72+
Mockito.doReturn(true).when(lockController).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt());
73+
74+
long start = System.currentTimeMillis();
75+
Assert.assertTrue(lockController.acquire(KEY, 30));
76+
long waited = System.currentTimeMillis() - start;
77+
78+
Assert.assertTrue("waited only " + waited + " ms", waited >= 4900);
79+
Mockito.verify(lockController, Mockito.times(2)).owns(KEY);
80+
Mockito.verify(lockController, Mockito.times(1)).doAcquire(Mockito.eq(KEY), Mockito.anyString(), Mockito.anyInt());
81+
}
82+
}

‎framework/jobs/src/main/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImpl.java‎

Lines changed: 45 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1288,7 +1288,7 @@ private void cleanupFailedSnapshotsCreatedWithDefaultStrategy(final long msid) {
12881288
* Cleanup networks that are stuck in Implementing state without associated async jobs.
12891289
* This only processes networks that have been stuck for longer than the job expiration threshold.
12901290
*/
1291-
private void cleanupOrphanedNetworks() {
1291+
protected void cleanupOrphanedNetworks() {
12921292
try {
12931293
SearchCriteria<NetworkVO> sc = networkDao.createSearchCriteria();
12941294
sc.addAnd("state", SearchCriteria.Op.EQ, Network.State.Implementing);
@@ -1313,30 +1313,54 @@ private void cleanupOrphanedNetworks() {
13131313
continue;
13141314
}
13151315

1316-
List<AsyncJobVO> jobs = _jobDao.findInstancePendingAsyncJobs("Network", network.getAccountId());
1317-
boolean hasActiveJob = false;
1318-
for (AsyncJobVO job : jobs) {
1319-
if (job.getInstanceId() != null && job.getInstanceId().equals(network.getId())) {
1320-
hasActiveJob = true;
1321-
break;
1322-
}
1323-
}
1316+
cleanupNetworkIfOrphaned(network.getId(), expireMinutes);
1317+
}
1318+
} catch (Exception e) {
1319+
logger.error("Error while cleaning up orphaned networks", e);
1320+
}
1321+
}
13241322

1325-
if (hasActiveJob) {
1326-
logger.debug("Network {} in Implementing state has active async job, skipping cleanup", network.getId());
1327-
continue;
1328-
}
1323+
/**
1324+
* The thread implementing a network holds the network's lock for the whole implementation, and that work can run
1325+
* inside a job of another type (a VM start, for example), so a network in Implementing is orphaned only if its lock
1326+
* is free. The lock is tried once without waiting, so the GC task does not block behind a running implementation,
1327+
* and the network is re-read under the lock before it is changed.
1328+
*/
1329+
private void cleanupNetworkIfOrphaned(long networkId, long expireMinutes) {
1330+
NetworkVO network = networkDao.acquireInLockTable(networkId, 0);
1331+
if (network == null) {
1332+
logger.debug("Network {} in Implementing state is locked (being implemented) or was removed, skipping cleanup", networkId);
1333+
return;
1334+
}
13291335

1330-
logger.warn("Found orphaned network {} in Implementing state without async job. " +
1331-
"Network created: {}, age: {} minutes, expiration threshold: {} minutes. Transitioning to Shutdown state.",
1332-
network.getId(), network.getCreated(),
1333-
(System.currentTimeMillis() - network.getCreated().getTime()) / 60000,
1334-
expireMinutes);
1335-
updateNetworkState(network);
1336+
try {
1337+
if (!Network.State.Implementing.equals(network.getState())) {
1338+
logger.debug("Network {} is no longer in Implementing state (now {}), skipping cleanup", networkId, network.getState());
1339+
return;
1340+
}
13361341

1342+
List<AsyncJobVO> jobs = _jobDao.findInstancePendingAsyncJobs("Network", network.getAccountId());
1343+
boolean hasActiveJob = false;
1344+
for (AsyncJobVO job : jobs) {
1345+
if (job.getInstanceId() != null && job.getInstanceId().equals(network.getId())) {
1346+
hasActiveJob = true;
1347+
break;
1348+
}
13371349
}
1338-
} catch (Exception e) {
1339-
logger.error("Error while cleaning up orphaned networks", e);
1350+
1351+
if (hasActiveJob) {
1352+
logger.debug("Network {} in Implementing state has active async job, skipping cleanup", network.getId());
1353+
return;
1354+
}
1355+
1356+
logger.warn("Found orphaned network {} in Implementing state without async job. " +
1357+
"Network created: {}, age: {} minutes, expiration threshold: {} minutes. Transitioning to Shutdown state.",
1358+
network.getId(), network.getCreated(),
1359+
(System.currentTimeMillis() - network.getCreated().getTime()) / 60000,
1360+
expireMinutes);
1361+
updateNetworkState(network);
1362+
} finally {
1363+
networkDao.releaseFromLockTable(networkId);
13401364
}
13411365
}
13421366

‎framework/jobs/src/test/java/org/apache/cloudstack/framework/jobs/impl/AsyncJobManagerImplTest.java‎

Lines changed: 125 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@
2121
import com.cloud.network.dao.NetworkDao;
2222
import com.cloud.network.dao.NetworkVO;
2323
import com.cloud.storage.Volume;
24+
import com.cloud.utils.db.SearchCriteria;
2425
import com.cloud.utils.fsm.NoTransitionException;
2526
import com.cloud.vm.VMInstanceVO;
2627
import com.cloud.vm.VirtualMachine;
@@ -30,14 +31,20 @@
3031
import org.apache.cloudstack.engine.orchestration.service.NetworkOrchestrationService;
3132
import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
3233
import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
34+
import org.apache.cloudstack.framework.jobs.dao.AsyncJobDao;
3335
import org.junit.Test;
3436
import org.junit.runner.RunWith;
37+
import org.mockito.InOrder;
3538
import org.mockito.InjectMocks;
3639
import org.mockito.Mock;
3740
import org.mockito.Mockito;
3841
import org.mockito.Spy;
3942
import org.mockito.junit.MockitoJUnitRunner;
4043

44+
import java.util.Collections;
45+
import java.util.Date;
46+
import java.util.List;
47+
4148
import static org.mockito.Mockito.when;
4249

4350
@RunWith(MockitoJUnitRunner.class)
@@ -55,6 +62,29 @@ public class AsyncJobManagerImplTest {
5562
NetworkDao networkDao;
5663
@Mock
5764
NetworkOrchestrationService networkOrchestrationService;
65+
@Mock
66+
AsyncJobDao asyncJobDao;
67+
68+
private static final long NETWORK_ID = 42L;
69+
private static final long ACCOUNT_ID = 2L;
70+
// job.expire.minutes defaults to one day; these networks were created two days ago
71+
private static final Date OLD_NETWORK_CREATED = new Date(System.currentTimeMillis() - 2 * 24 * 60 * 60 * 1000L);
72+
73+
private NetworkVO mockImplementingNetworkListed(Date created) {
74+
NetworkVO listed = Mockito.mock(NetworkVO.class);
75+
when(listed.getCreated()).thenReturn(created);
76+
SearchCriteria<NetworkVO> sc = Mockito.mock(SearchCriteria.class);
77+
when(networkDao.createSearchCriteria()).thenReturn(sc);
78+
when(networkDao.search(sc, null)).thenReturn(List.of(listed));
79+
return listed;
80+
}
81+
82+
private NetworkVO mockNetworkReadUnderLock(Network.State state) {
83+
NetworkVO locked = Mockito.mock(NetworkVO.class);
84+
when(locked.getState()).thenReturn(state);
85+
when(networkDao.acquireInLockTable(NETWORK_ID, 0)).thenReturn(locked);
86+
return locked;
87+
}
5888

5989
@Test
6090
public void testCleanupVolumeResource() {
@@ -93,4 +123,99 @@ public void testCleanupNetworkResource() throws NoTransitionException {
93123
Mockito.verify(networkOrchestrationService, Mockito.times(1)).stateTransitTo(networkVO,
94124
Network.Event.OperationFailed);
95125
}
126+
127+
@Test
128+
public void testCleanupOrphanedNetworksSkipsNetworkWhoseLockIsHeld() throws NoTransitionException {
129+
NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED);
130+
when(listed.getId()).thenReturn(NETWORK_ID);
131+
// a VM start implementing the network holds its lock, so a try without waiting fails
132+
when(networkDao.acquireInLockTable(NETWORK_ID, 0)).thenReturn(null);
133+
134+
asyncJobManager.cleanupOrphanedNetworks();
135+
136+
Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any());
137+
Mockito.verify(networkDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any());
138+
Mockito.verify(networkDao, Mockito.never()).releaseFromLockTable(Mockito.anyLong());
139+
Mockito.verifyNoInteractions(asyncJobDao);
140+
}
141+
142+
@Test
143+
public void testCleanupOrphanedNetworksShutsDownOrphanedNetworkUnderLock() throws NoTransitionException {
144+
NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED);
145+
when(listed.getId()).thenReturn(NETWORK_ID);
146+
NetworkVO locked = mockNetworkReadUnderLock(Network.State.Implementing);
147+
when(locked.getId()).thenReturn(NETWORK_ID);
148+
when(locked.getAccountId()).thenReturn(ACCOUNT_ID);
149+
when(locked.getCreated()).thenReturn(OLD_NETWORK_CREATED);
150+
when(asyncJobDao.findInstancePendingAsyncJobs("Network", ACCOUNT_ID)).thenReturn(Collections.emptyList());
151+
152+
asyncJobManager.cleanupOrphanedNetworks();
153+
154+
InOrder inOrder = Mockito.inOrder(networkDao, networkOrchestrationService);
155+
inOrder.verify(networkDao).acquireInLockTable(NETWORK_ID, 0);
156+
inOrder.verify(networkOrchestrationService).stateTransitTo(locked, Network.Event.OperationFailed);
157+
inOrder.verify(networkDao).releaseFromLockTable(NETWORK_ID);
158+
Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.eq(listed), Mockito.any());
159+
}
160+
161+
@Test
162+
public void testCleanupOrphanedNetworksSkipsNetworkImplementedBeforeItWasLocked() throws NoTransitionException {
163+
NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED);
164+
when(listed.getId()).thenReturn(NETWORK_ID);
165+
// the implementation finished between the search and the lock
166+
mockNetworkReadUnderLock(Network.State.Implemented);
167+
168+
asyncJobManager.cleanupOrphanedNetworks();
169+
170+
Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any());
171+
Mockito.verify(networkDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any());
172+
Mockito.verify(networkDao).releaseFromLockTable(NETWORK_ID);
173+
}
174+
175+
@Test
176+
public void testCleanupOrphanedNetworksSkipsNetworkWithPendingNetworkJob() throws NoTransitionException {
177+
NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED);
178+
when(listed.getId()).thenReturn(NETWORK_ID);
179+
NetworkVO locked = mockNetworkReadUnderLock(Network.State.Implementing);
180+
when(locked.getId()).thenReturn(NETWORK_ID);
181+
when(locked.getAccountId()).thenReturn(ACCOUNT_ID);
182+
AsyncJobVO job = new AsyncJobVO();
183+
job.setInstanceId(NETWORK_ID);
184+
when(asyncJobDao.findInstancePendingAsyncJobs("Network", ACCOUNT_ID)).thenReturn(List.of(job));
185+
186+
asyncJobManager.cleanupOrphanedNetworks();
187+
188+
Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any());
189+
Mockito.verify(networkDao).releaseFromLockTable(NETWORK_ID);
190+
}
191+
192+
@Test
193+
public void testCleanupOrphanedNetworksSkipsYoungNetwork() throws NoTransitionException {
194+
mockImplementingNetworkListed(new Date());
195+
196+
asyncJobManager.cleanupOrphanedNetworks();
197+
198+
Mockito.verify(networkDao, Mockito.never()).acquireInLockTable(Mockito.anyLong(), Mockito.anyInt());
199+
Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any());
200+
Mockito.verifyNoInteractions(asyncJobDao);
201+
}
202+
203+
@Test
204+
public void testCleanupOrphanedNetworksReleasesLockWhenForcingShutdown() throws NoTransitionException {
205+
NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED);
206+
when(listed.getId()).thenReturn(NETWORK_ID);
207+
NetworkVO locked = mockNetworkReadUnderLock(Network.State.Implementing);
208+
when(locked.getId()).thenReturn(NETWORK_ID);
209+
when(locked.getAccountId()).thenReturn(ACCOUNT_ID);
210+
when(locked.getCreated()).thenReturn(OLD_NETWORK_CREATED);
211+
when(asyncJobDao.findInstancePendingAsyncJobs("Network", ACCOUNT_ID)).thenReturn(Collections.emptyList());
212+
when(networkOrchestrationService.stateTransitTo(locked, Network.Event.OperationFailed)).thenThrow(new NoTransitionException("no transition"));
213+
214+
asyncJobManager.cleanupOrphanedNetworks();
215+
216+
InOrder inOrder = Mockito.inOrder(locked, networkDao);
217+
inOrder.verify(locked).setState(Network.State.Shutdown);
218+
inOrder.verify(networkDao).update(NETWORK_ID, locked);
219+
inOrder.verify(networkDao).releaseFromLockTable(NETWORK_ID);
220+
}
96221
}

0 commit comments

Comments
 (0)