Skip to content

Commit 4cea137

Browse files
Dogface2kDogface2k
committed
NSX: harden VPN status polling
1 parent 8d073eb commit 4cea137

5 files changed

Lines changed: 229 additions & 14 deletions

File tree

‎engine/schema/src/main/java/com/cloud/network/dao/Site2SiteVpnConnectionDao.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818

1919
import java.util.List;
2020

21+
import com.cloud.network.Site2SiteVpnConnection;
2122
import com.cloud.utils.db.GenericDao;
2223

2324
public interface Site2SiteVpnConnectionDao extends GenericDao<Site2SiteVpnConnectionVO, Long> {
@@ -27,6 +28,8 @@ public interface Site2SiteVpnConnectionDao extends GenericDao<Site2SiteVpnConnec
2728

2829
List<Site2SiteVpnConnectionVO> listByVpcId(long vpcId);
2930

31+
List<Site2SiteVpnConnectionVO> listByStates(Site2SiteVpnConnection.State... states);
32+
3033
Site2SiteVpnConnectionVO findByVpnGatewayIdAndCustomerGatewayId(long vpnId, long customerId);
3134

3235
Site2SiteVpnConnectionVO findByCustomerGatewayId(long customerId);

‎engine/schema/src/main/java/com/cloud/network/dao/Site2SiteVpnConnectionDaoImpl.java‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323

2424
import org.springframework.stereotype.Component;
2525

26+
import com.cloud.network.Site2SiteVpnConnection;
2627
import com.cloud.utils.db.GenericDaoBase;
2728
import com.cloud.utils.db.JoinBuilder.JoinType;
2829
import com.cloud.utils.db.SearchBuilder;
@@ -39,6 +40,7 @@ public class Site2SiteVpnConnectionDaoImpl extends GenericDaoBase<Site2SiteVpnCo
3940
private SearchBuilder<Site2SiteVpnConnectionVO> AllFieldsSearch;
4041
private SearchBuilder<Site2SiteVpnConnectionVO> VpcSearch;
4142
private SearchBuilder<Site2SiteVpnGatewayVO> VpnGatewaySearch;
43+
private SearchBuilder<Site2SiteVpnConnectionVO> StateSearch;
4244

4345
public Site2SiteVpnConnectionDaoImpl() {
4446
}
@@ -55,6 +57,10 @@ protected void init() {
5557
VpnGatewaySearch.and("vpcId", VpnGatewaySearch.entity().getVpcId(), SearchCriteria.Op.EQ);
5658
VpcSearch.join("vpnGatewaySearch", VpnGatewaySearch, VpnGatewaySearch.entity().getId(), VpcSearch.entity().getVpnGatewayId(), JoinType.INNER);
5759
VpcSearch.done();
60+
61+
StateSearch = createSearchBuilder();
62+
StateSearch.and("state", StateSearch.entity().getState(), SearchCriteria.Op.IN);
63+
StateSearch.done();
5864
}
5965

6066
@Override
@@ -78,6 +84,13 @@ public List<Site2SiteVpnConnectionVO> listByVpcId(long vpcId) {
7884
return listBy(sc);
7985
}
8086

87+
@Override
88+
public List<Site2SiteVpnConnectionVO> listByStates(Site2SiteVpnConnection.State... states) {
89+
SearchCriteria<Site2SiteVpnConnectionVO> sc = StateSearch.create();
90+
sc.setParameters("state", (Object[]) states);
91+
return listBy(sc);
92+
}
93+
8194
@Override
8295
public Site2SiteVpnConnectionVO findByVpnGatewayIdAndCustomerGatewayId(long vpnId, long customerId) {
8396
SearchCriteria<Site2SiteVpnConnectionVO> sc = AllFieldsSearch.create();

‎engine/schema/src/test/java/com/cloud/network/dao/Site2SiteVpnConnectionDaoImplTest.java‎

Lines changed: 63 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,63 @@
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.network.dao;
18+
19+
import java.util.List;
20+
21+
import org.junit.Test;
22+
import org.junit.runner.RunWith;
23+
import org.mockito.Mock;
24+
import org.mockito.Spy;
25+
import org.mockito.junit.MockitoJUnitRunner;
26+
import org.springframework.test.util.ReflectionTestUtils;
27+
28+
import com.cloud.network.Site2SiteVpnConnection;
29+
import com.cloud.utils.db.SearchBuilder;
30+
import com.cloud.utils.db.SearchCriteria;
31+
32+
import static org.mockito.Mockito.doReturn;
33+
import static org.mockito.Mockito.verify;
34+
import static org.mockito.Mockito.when;
35+
36+
@RunWith(MockitoJUnitRunner.class)
37+
public class Site2SiteVpnConnectionDaoImplTest {
38+
39+
@Spy
40+
private Site2SiteVpnConnectionDaoImpl dao;
41+
@Mock
42+
private SearchBuilder<Site2SiteVpnConnectionVO> stateSearch;
43+
@Mock
44+
private SearchCriteria<Site2SiteVpnConnectionVO> searchCriteria;
45+
46+
@Test
47+
public void testListByStatesUsesStateSearchCriteria() {
48+
ReflectionTestUtils.setField(dao, "StateSearch", stateSearch);
49+
when(stateSearch.create()).thenReturn(searchCriteria);
50+
doReturn(List.of()).when(dao).listBy(searchCriteria);
51+
Site2SiteVpnConnection.State[] states = {
52+
Site2SiteVpnConnection.State.Pending,
53+
Site2SiteVpnConnection.State.Connecting,
54+
Site2SiteVpnConnection.State.Connected,
55+
Site2SiteVpnConnection.State.Disconnected
56+
};
57+
58+
dao.listByStates(states);
59+
60+
verify(searchCriteria).setParameters("state", (Object[]) states);
61+
verify(dao).listBy(searchCriteria);
62+
}
63+
}

‎plugins/network-elements/nsx/src/main/java/org/apache/cloudstack/service/NsxServiceImpl.java‎

Lines changed: 22 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -116,15 +116,14 @@ public class NsxServiceImpl extends ManagerBase implements NsxService, Configura
116116
@Override
117117
public boolean configure(String name, Map<String, Object> params) throws ConfigurationException {
118118
super.configure(name, params);
119-
vpnStatusPollExecutor = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Nsx-Vpn-Status-Poll"));
120119
return true;
121120
}
122121

123122
@Override
124-
public boolean start() {
123+
public synchronized boolean start() {
125124
super.start();
126-
if (vpnStatusPollExecutor == null) {
127-
throw new IllegalStateException("NSX VPN status poller was not configured");
125+
if (vpnStatusPollExecutor != null && !vpnStatusPollExecutor.isShutdown()) {
126+
return true;
128127
}
129128
Integer configuredInterval = NSX_VPN_STATUS_POLL_INTERVAL.value();
130129
int pollInterval = Objects.isNull(configuredInterval) ? VPN_STATUS_POLL_DEFAULT_INTERVAL : configuredInterval;
@@ -133,14 +132,27 @@ public boolean start() {
133132
configuredInterval, NSX_VPN_STATUS_POLL_INTERVAL.key(), VPN_STATUS_POLL_MIN_INTERVAL, VPN_STATUS_POLL_DEFAULT_INTERVAL);
134133
pollInterval = VPN_STATUS_POLL_DEFAULT_INTERVAL;
135134
}
136-
vpnStatusPollExecutor.scheduleWithFixedDelay(new VpnStatusPollTask(), pollInterval, pollInterval, TimeUnit.SECONDS);
135+
ScheduledExecutorService executor = createVpnStatusPollExecutor();
136+
try {
137+
executor.scheduleWithFixedDelay(new VpnStatusPollTask(), pollInterval, pollInterval, TimeUnit.SECONDS);
138+
vpnStatusPollExecutor = executor;
139+
} catch (RuntimeException e) {
140+
executor.shutdownNow();
141+
throw e;
142+
}
137143
return true;
138144
}
139145

146+
protected ScheduledExecutorService createVpnStatusPollExecutor() {
147+
return Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("Nsx-Vpn-Status-Poll"));
148+
}
149+
140150
@Override
141-
public boolean stop() {
142-
if (Objects.nonNull(vpnStatusPollExecutor)) {
143-
vpnStatusPollExecutor.shutdownNow();
151+
public synchronized boolean stop() {
152+
ScheduledExecutorService executor = vpnStatusPollExecutor;
153+
vpnStatusPollExecutor = null;
154+
if (Objects.nonNull(executor)) {
155+
executor.shutdownNow();
144156
}
145157
return super.stop();
146158
}
@@ -339,11 +351,9 @@ protected class VpnStatusPollTask extends ManagedContextRunnable {
339351
protected void runInContext() {
340352
try {
341353
Set<Long> polledConnectionIds = new HashSet<>();
342-
List<Site2SiteVpnConnectionVO> connections = site2SiteVpnConnectionDao.listAll();
354+
List<Site2SiteVpnConnectionVO> connections = site2SiteVpnConnectionDao.listByStates(
355+
VPN_POLLED_STATES.toArray(new Site2SiteVpnConnection.State[0]));
343356
for (Site2SiteVpnConnectionVO connection : connections) {
344-
if (!VPN_POLLED_STATES.contains(connection.getState())) {
345-
continue;
346-
}
347357
Site2SiteVpnGatewayVO vpnGateway = site2SiteVpnGatewayDao.findById(connection.getVpnGatewayId());
348358
if (vpnGateway == null) {
349359
continue;

‎plugins/network-elements/nsx/src/test/java/org/apache/cloudstack/service/NsxServiceImplTest.java‎

Lines changed: 128 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -50,7 +50,10 @@
5050
import org.mockito.junit.MockitoJUnitRunner;
5151

5252
import java.util.List;
53+
import java.util.Map;
54+
import java.util.concurrent.ScheduledExecutorService;
5355
import java.util.concurrent.atomic.AtomicBoolean;
56+
import java.util.concurrent.atomic.AtomicInteger;
5457
import java.util.concurrent.atomic.AtomicReference;
5558

5659
import static org.junit.Assert.assertEquals;
@@ -62,6 +65,7 @@
6265
import static org.mockito.Mockito.mock;
6366
import static org.mockito.Mockito.doNothing;
6467
import static org.mockito.Mockito.never;
68+
import static org.mockito.Mockito.times;
6569
import static org.mockito.Mockito.verify;
6670
import static org.mockito.Mockito.when;
6771

@@ -233,6 +237,84 @@ protected void transitionVpnConnectionState(Site2SiteVpnConnectionVO connection,
233237
assertEquals(Site2SiteVpnConnection.State.Connected, transitionedState.get());
234238
}
235239

240+
@Test
241+
public void testPollVpnConnectionStatusTransitionsDown() {
242+
Site2SiteVpnConnectionVO connection = mock(Site2SiteVpnConnectionVO.class);
243+
VpcVO vpc = mock(VpcVO.class);
244+
when(connection.getState()).thenReturn(Site2SiteVpnConnection.State.Connected);
245+
AtomicReference<Site2SiteVpnConnection.State> transitionedState = new AtomicReference<>();
246+
247+
NsxServiceImpl service = new NsxServiceImpl() {
248+
@Override
249+
public String getVpnConnectionStatus(Vpc vpc, String connectionUuid) {
250+
return VPN_SESSION_STATUS_DOWN;
251+
}
252+
253+
@Override
254+
protected void transitionVpnConnectionState(Site2SiteVpnConnectionVO connection, VpcVO vpc,
255+
Site2SiteVpnConnection.State observedState,
256+
Site2SiteVpnConnection.State newState) {
257+
transitionedState.set(newState);
258+
}
259+
};
260+
261+
service.pollVpnConnectionStatus(connection, vpc);
262+
263+
assertEquals(Site2SiteVpnConnection.State.Disconnected, transitionedState.get());
264+
}
265+
266+
@Test
267+
public void testPollVpnConnectionStatusKeepsPendingConnectionWhenSessionIsNotFound() {
268+
Site2SiteVpnConnectionVO connection = mock(Site2SiteVpnConnectionVO.class);
269+
VpcVO vpc = mock(VpcVO.class);
270+
when(connection.getState()).thenReturn(Site2SiteVpnConnection.State.Pending);
271+
AtomicBoolean transitioned = new AtomicBoolean();
272+
273+
NsxServiceImpl service = new NsxServiceImpl() {
274+
@Override
275+
public String getVpnConnectionStatus(Vpc vpc, String connectionUuid) {
276+
return VPN_SESSION_STATUS_NOT_FOUND;
277+
}
278+
279+
@Override
280+
protected void transitionVpnConnectionState(Site2SiteVpnConnectionVO connection, VpcVO vpc,
281+
Site2SiteVpnConnection.State observedState,
282+
Site2SiteVpnConnection.State newState) {
283+
transitioned.set(true);
284+
}
285+
};
286+
287+
service.pollVpnConnectionStatus(connection, vpc);
288+
289+
assertFalse(transitioned.get());
290+
}
291+
292+
@Test
293+
public void testPollVpnConnectionStatusMarksMissingConnectedSessionAsError() {
294+
Site2SiteVpnConnectionVO connection = mock(Site2SiteVpnConnectionVO.class);
295+
VpcVO vpc = mock(VpcVO.class);
296+
when(connection.getState()).thenReturn(Site2SiteVpnConnection.State.Connected);
297+
AtomicReference<Site2SiteVpnConnection.State> transitionedState = new AtomicReference<>();
298+
299+
NsxServiceImpl service = new NsxServiceImpl() {
300+
@Override
301+
public String getVpnConnectionStatus(Vpc vpc, String connectionUuid) {
302+
return VPN_SESSION_STATUS_NOT_FOUND;
303+
}
304+
305+
@Override
306+
protected void transitionVpnConnectionState(Site2SiteVpnConnectionVO connection, VpcVO vpc,
307+
Site2SiteVpnConnection.State observedState,
308+
Site2SiteVpnConnection.State newState) {
309+
transitionedState.set(newState);
310+
}
311+
};
312+
313+
service.pollVpnConnectionStatus(connection, vpc);
314+
315+
assertEquals(Site2SiteVpnConnection.State.Error, transitionedState.get());
316+
}
317+
236318
@Test
237319
public void testPollVpnConnectionStatusDoesNotTransitionOnQueryFailure() {
238320
Site2SiteVpnConnectionVO connection = mock(Site2SiteVpnConnectionVO.class);
@@ -308,12 +390,56 @@ public void testVpnStatusPollerUsesPersistedOwnershipAfterOfferingChanges() {
308390
verify(service).pollVpnConnectionStatus(connection, vpc);
309391
}
310392

393+
@Test
394+
public void testVpnStatusPollerQueriesOnlyPollableStates() {
395+
nsxService.new VpnStatusPollTask().runInContext();
396+
397+
verify(site2SiteVpnConnectionDao).listByStates(
398+
Site2SiteVpnConnection.State.Pending,
399+
Site2SiteVpnConnection.State.Connecting,
400+
Site2SiteVpnConnection.State.Connected,
401+
Site2SiteVpnConnection.State.Disconnected);
402+
verify(site2SiteVpnConnectionDao, never()).listAll();
403+
}
404+
405+
@Test
406+
public void testVpnStatusPollerCanRestartInSameJvm() throws Exception {
407+
ScheduledExecutorService firstExecutor = mock(ScheduledExecutorService.class);
408+
ScheduledExecutorService secondExecutor = mock(ScheduledExecutorService.class);
409+
AtomicInteger executorIndex = new AtomicInteger();
410+
NsxServiceImpl service = new NsxServiceImpl() {
411+
@Override
412+
protected ScheduledExecutorService createVpnStatusPollExecutor() {
413+
return executorIndex.getAndIncrement() == 0 ? firstExecutor : secondExecutor;
414+
}
415+
};
416+
service.configure("NsxService", Map.of());
417+
try {
418+
assertTrue(service.start());
419+
verify(firstExecutor).scheduleWithFixedDelay(any(Runnable.class), eq(60L), eq(60L), eq(java.util.concurrent.TimeUnit.SECONDS));
420+
421+
assertTrue(service.stop());
422+
verify(firstExecutor).shutdownNow();
423+
424+
assertTrue(service.start());
425+
verify(secondExecutor).scheduleWithFixedDelay(any(Runnable.class), eq(60L), eq(60L), eq(java.util.concurrent.TimeUnit.SECONDS));
426+
assertEquals(2, executorIndex.get());
427+
} finally {
428+
service.stop();
429+
}
430+
verify(secondExecutor).shutdownNow();
431+
verify(firstExecutor, times(1)).shutdownNow();
432+
}
433+
311434
private Site2SiteVpnConnectionVO mockPollableVpnConnection() {
312435
Site2SiteVpnConnectionVO connection = mock(Site2SiteVpnConnectionVO.class);
313436
when(connection.getId()).thenReturn(11L);
314437
when(connection.getVpnGatewayId()).thenReturn(7L);
315-
when(connection.getState()).thenReturn(Site2SiteVpnConnection.State.Connected);
316-
when(site2SiteVpnConnectionDao.listAll()).thenReturn(List.of(connection));
438+
when(site2SiteVpnConnectionDao.listByStates(
439+
Site2SiteVpnConnection.State.Pending,
440+
Site2SiteVpnConnection.State.Connecting,
441+
Site2SiteVpnConnection.State.Connected,
442+
Site2SiteVpnConnection.State.Disconnected)).thenReturn(List.of(connection));
317443
return connection;
318444
}
319445

0 commit comments

Comments
 (0)