Skip to content

Commit d02adcd

Browse files
sureshanapartisureshanaparti
committed
Merge branch '4.20' into 4.22
2 parents 56306ab + 2ebcaf6 commit d02adcd

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
@@ -1364,7 +1364,7 @@ private void cleanupFailedSnapshotsCreatedWithDefaultStrategy(final long msid) {
13641364
* Cleanup networks that are stuck in Implementing state without associated async jobs.
13651365
* This only processes networks that have been stuck for longer than the job expiration threshold.
13661366
*/
1367-
private void cleanupOrphanedNetworks() {
1367+
protected void cleanupOrphanedNetworks() {
13681368
try {
13691369
SearchCriteria<NetworkVO> sc = networkDao.createSearchCriteria();
13701370
sc.addAnd("state", SearchCriteria.Op.EQ, Network.State.Implementing);
@@ -1389,30 +1389,54 @@ private void cleanupOrphanedNetworks() {
13891389
continue;
13901390
}
13911391

1392-
List<AsyncJobVO> jobs = _jobDao.findInstancePendingAsyncJobs("Network", network.getAccountId());
1393-
boolean hasActiveJob = false;
1394-
for (AsyncJobVO job : jobs) {
1395-
if (job.getInstanceId() != null && job.getInstanceId().equals(network.getId())) {
1396-
hasActiveJob = true;
1397-
break;
1398-
}
1399-
}
1392+
cleanupNetworkIfOrphaned(network.getId(), expireMinutes);
1393+
}
1394+
} catch (Exception e) {
1395+
logger.error("Error while cleaning up orphaned networks", e);
1396+
}
1397+
}
14001398

1401-
if (hasActiveJob) {
1402-
logger.debug("Network {} in Implementing state has active async job, skipping cleanup", network.getId());
1403-
continue;
1404-
}
1399+
/**
1400+
* The thread implementing a network holds the network's lock for the whole implementation, and that work can run
1401+
* inside a job of another type (a VM start, for example), so a network in Implementing is orphaned only if its lock
1402+
* is free. The lock is tried once without waiting, so the GC task does not block behind a running implementation,
1403+
* and the network is re-read under the lock before it is changed.
1404+
*/
1405+
private void cleanupNetworkIfOrphaned(long networkId, long expireMinutes) {
1406+
NetworkVO network = networkDao.acquireInLockTable(networkId, 0);
1407+
if (network == null) {
1408+
logger.debug("Network {} in Implementing state is locked (being implemented) or was removed, skipping cleanup", networkId);
1409+
return;
1410+
}
14051411

1406-
logger.warn("Found orphaned network {} in Implementing state without async job. " +
1407-
"Network created: {}, age: {} minutes, expiration threshold: {} minutes. Transitioning to Shutdown state.",
1408-
network.getId(), network.getCreated(),
1409-
(System.currentTimeMillis() - network.getCreated().getTime()) / 60000,
1410-
expireMinutes);
1411-
updateNetworkState(network);
1412+
try {
1413+
if (!Network.State.Implementing.equals(network.getState())) {
1414+
logger.debug("Network {} is no longer in Implementing state (now {}), skipping cleanup", networkId, network.getState());
1415+
return;
1416+
}
14121417

1418+
List<AsyncJobVO> jobs = _jobDao.findInstancePendingAsyncJobs("Network", network.getAccountId());
1419+
boolean hasActiveJob = false;
1420+
for (AsyncJobVO job : jobs) {
1421+
if (job.getInstanceId() != null && job.getInstanceId().equals(network.getId())) {
1422+
hasActiveJob = true;
1423+
break;
1424+
}
14131425
}
1414-
} catch (Exception e) {
1415-
logger.error("Error while cleaning up orphaned networks", e);
1426+
1427+
if (hasActiveJob) {
1428+
logger.debug("Network {} in Implementing state has active async job, skipping cleanup", network.getId());
1429+
return;
1430+
}
1431+
1432+
logger.warn("Found orphaned network {} in Implementing state without async job. " +
1433+
"Network created: {}, age: {} minutes, expiration threshold: {} minutes. Transitioning to Shutdown state.",
1434+
network.getId(), network.getCreated(),
1435+
(System.currentTimeMillis() - network.getCreated().getTime()) / 60000,
1436+
expireMinutes);
1437+
updateNetworkState(network);
1438+
} finally {
1439+
networkDao.releaseFromLockTable(networkId);
14161440
}
14171441
}
14181442

‎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
@@ -27,6 +27,7 @@
2727
import com.cloud.network.dao.NetworkVO;
2828
import com.cloud.storage.Volume;
2929
import com.cloud.utils.db.DbProperties;
30+
import com.cloud.utils.db.SearchCriteria;
3031
import com.cloud.utils.fsm.NoTransitionException;
3132
import com.cloud.vm.VMInstanceVO;
3233
import com.cloud.vm.VirtualMachine;
@@ -38,17 +39,23 @@
3839
import org.apache.cloudstack.engine.subsystem.api.storage.VolumeDataFactory;
3940
import org.apache.cloudstack.engine.subsystem.api.storage.VolumeInfo;
4041
import org.apache.cloudstack.framework.config.ConfigKey;
42+
import org.apache.cloudstack.framework.jobs.dao.AsyncJobDao;
4143
import org.junit.After;
4244
import org.junit.Assert;
4345
import org.junit.Before;
4446
import org.junit.Test;
4547
import org.junit.runner.RunWith;
48+
import org.mockito.InOrder;
4649
import org.mockito.InjectMocks;
4750
import org.mockito.Mock;
4851
import org.mockito.Mockito;
4952
import org.mockito.Spy;
5053
import org.mockito.junit.MockitoJUnitRunner;
5154

55+
import java.util.Collections;
56+
import java.util.Date;
57+
import java.util.List;
58+
5259
import static org.mockito.Mockito.when;
5360

5461
@RunWith(MockitoJUnitRunner.class)
@@ -72,6 +79,29 @@ public class AsyncJobManagerImplTest {
7279
NetworkDao networkDao;
7380
@Mock
7481
NetworkOrchestrationService networkOrchestrationService;
82+
@Mock
83+
AsyncJobDao asyncJobDao;
84+
85+
private static final long NETWORK_ID = 42L;
86+
private static final long ACCOUNT_ID = 2L;
87+
// job.expire.minutes defaults to one day; these networks were created two days ago
88+
private static final Date OLD_NETWORK_CREATED = new Date(System.currentTimeMillis() - 2 * 24 * 60 * 60 * 1000L);
89+
90+
private NetworkVO mockImplementingNetworkListed(Date created) {
91+
NetworkVO listed = Mockito.mock(NetworkVO.class);
92+
when(listed.getCreated()).thenReturn(created);
93+
SearchCriteria<NetworkVO> sc = Mockito.mock(SearchCriteria.class);
94+
when(networkDao.createSearchCriteria()).thenReturn(sc);
95+
when(networkDao.search(sc, null)).thenReturn(List.of(listed));
96+
return listed;
97+
}
98+
99+
private NetworkVO mockNetworkReadUnderLock(Network.State state) {
100+
NetworkVO locked = Mockito.mock(NetworkVO.class);
101+
when(locked.getState()).thenReturn(state);
102+
when(networkDao.acquireInLockTable(NETWORK_ID, 0)).thenReturn(locked);
103+
return locked;
104+
}
75105

76106
@Before
77107
public void preloadDbProperties() throws Exception {
@@ -135,6 +165,101 @@ public void testCleanupNetworkResource() throws NoTransitionException {
135165
Network.Event.OperationFailed);
136166
}
137167

168+
@Test
169+
public void testCleanupOrphanedNetworksSkipsNetworkWhoseLockIsHeld() throws NoTransitionException {
170+
NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED);
171+
when(listed.getId()).thenReturn(NETWORK_ID);
172+
// a VM start implementing the network holds its lock, so a try without waiting fails
173+
when(networkDao.acquireInLockTable(NETWORK_ID, 0)).thenReturn(null);
174+
175+
asyncJobManager.cleanupOrphanedNetworks();
176+
177+
Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any());
178+
Mockito.verify(networkDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any());
179+
Mockito.verify(networkDao, Mockito.never()).releaseFromLockTable(Mockito.anyLong());
180+
Mockito.verifyNoInteractions(asyncJobDao);
181+
}
182+
183+
@Test
184+
public void testCleanupOrphanedNetworksShutsDownOrphanedNetworkUnderLock() throws NoTransitionException {
185+
NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED);
186+
when(listed.getId()).thenReturn(NETWORK_ID);
187+
NetworkVO locked = mockNetworkReadUnderLock(Network.State.Implementing);
188+
when(locked.getId()).thenReturn(NETWORK_ID);
189+
when(locked.getAccountId()).thenReturn(ACCOUNT_ID);
190+
when(locked.getCreated()).thenReturn(OLD_NETWORK_CREATED);
191+
when(asyncJobDao.findInstancePendingAsyncJobs("Network", ACCOUNT_ID)).thenReturn(Collections.emptyList());
192+
193+
asyncJobManager.cleanupOrphanedNetworks();
194+
195+
InOrder inOrder = Mockito.inOrder(networkDao, networkOrchestrationService);
196+
inOrder.verify(networkDao).acquireInLockTable(NETWORK_ID, 0);
197+
inOrder.verify(networkOrchestrationService).stateTransitTo(locked, Network.Event.OperationFailed);
198+
inOrder.verify(networkDao).releaseFromLockTable(NETWORK_ID);
199+
Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.eq(listed), Mockito.any());
200+
}
201+
202+
@Test
203+
public void testCleanupOrphanedNetworksSkipsNetworkImplementedBeforeItWasLocked() throws NoTransitionException {
204+
NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED);
205+
when(listed.getId()).thenReturn(NETWORK_ID);
206+
// the implementation finished between the search and the lock
207+
mockNetworkReadUnderLock(Network.State.Implemented);
208+
209+
asyncJobManager.cleanupOrphanedNetworks();
210+
211+
Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any());
212+
Mockito.verify(networkDao, Mockito.never()).update(Mockito.anyLong(), Mockito.any());
213+
Mockito.verify(networkDao).releaseFromLockTable(NETWORK_ID);
214+
}
215+
216+
@Test
217+
public void testCleanupOrphanedNetworksSkipsNetworkWithPendingNetworkJob() throws NoTransitionException {
218+
NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED);
219+
when(listed.getId()).thenReturn(NETWORK_ID);
220+
NetworkVO locked = mockNetworkReadUnderLock(Network.State.Implementing);
221+
when(locked.getId()).thenReturn(NETWORK_ID);
222+
when(locked.getAccountId()).thenReturn(ACCOUNT_ID);
223+
AsyncJobVO job = new AsyncJobVO();
224+
job.setInstanceId(NETWORK_ID);
225+
when(asyncJobDao.findInstancePendingAsyncJobs("Network", ACCOUNT_ID)).thenReturn(List.of(job));
226+
227+
asyncJobManager.cleanupOrphanedNetworks();
228+
229+
Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any());
230+
Mockito.verify(networkDao).releaseFromLockTable(NETWORK_ID);
231+
}
232+
233+
@Test
234+
public void testCleanupOrphanedNetworksSkipsYoungNetwork() throws NoTransitionException {
235+
mockImplementingNetworkListed(new Date());
236+
237+
asyncJobManager.cleanupOrphanedNetworks();
238+
239+
Mockito.verify(networkDao, Mockito.never()).acquireInLockTable(Mockito.anyLong(), Mockito.anyInt());
240+
Mockito.verify(networkOrchestrationService, Mockito.never()).stateTransitTo(Mockito.any(), Mockito.any());
241+
Mockito.verifyNoInteractions(asyncJobDao);
242+
}
243+
244+
@Test
245+
public void testCleanupOrphanedNetworksReleasesLockWhenForcingShutdown() throws NoTransitionException {
246+
NetworkVO listed = mockImplementingNetworkListed(OLD_NETWORK_CREATED);
247+
when(listed.getId()).thenReturn(NETWORK_ID);
248+
NetworkVO locked = mockNetworkReadUnderLock(Network.State.Implementing);
249+
when(locked.getId()).thenReturn(NETWORK_ID);
250+
when(locked.getAccountId()).thenReturn(ACCOUNT_ID);
251+
when(locked.getCreated()).thenReturn(OLD_NETWORK_CREATED);
252+
when(asyncJobDao.findInstancePendingAsyncJobs("Network", ACCOUNT_ID)).thenReturn(Collections.emptyList());
253+
when(networkOrchestrationService.stateTransitTo(locked, Network.Event.OperationFailed)).thenThrow(new NoTransitionException("no transition"));
254+
255+
asyncJobManager.cleanupOrphanedNetworks();
256+
257+
InOrder inOrder = Mockito.inOrder(locked, networkDao);
258+
inOrder.verify(locked).setState(Network.State.Shutdown);
259+
inOrder.verify(networkDao).update(NETWORK_ID, locked);
260+
inOrder.verify(networkDao).releaseFromLockTable(NETWORK_ID);
261+
}
262+
138263
@Test
139264
public void configureApiPoolUsesConfiguredValueWhenLargerThanDbDerived() throws Exception {
140265
overrideConfigValue(AsyncJobManagerImpl.ApiJobPoolSize, 3000);

0 commit comments

Comments
 (0)