Improve NETCONF session ID handling
[netconf.git] / protocol / netconf-server / src / test / java / org / opendaylight / netconf / server / ConcurrentClientsTest.java
1 /*
2  * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
3  *
4  * This program and the accompanying materials are made available under the
5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
6  * and is available at http://www.eclipse.org/legal/epl-v10.html
7  */
8 package org.opendaylight.netconf.server;
9
10 import static com.google.common.base.Preconditions.checkState;
11 import static java.util.Objects.requireNonNull;
12 import static org.junit.Assert.assertEquals;
13 import static org.mockito.ArgumentMatchers.any;
14 import static org.mockito.Mockito.doNothing;
15 import static org.mockito.Mockito.doReturn;
16 import static org.mockito.Mockito.mock;
17
18 import com.google.common.io.ByteStreams;
19 import io.netty.channel.ChannelFuture;
20 import io.netty.channel.EventLoopGroup;
21 import io.netty.channel.nio.NioEventLoopGroup;
22 import io.netty.util.HashedWheelTimer;
23 import io.netty.util.concurrent.GlobalEventExecutor;
24 import java.io.DataOutputStream;
25 import java.io.InputStream;
26 import java.io.InputStreamReader;
27 import java.net.InetSocketAddress;
28 import java.net.Socket;
29 import java.util.ArrayList;
30 import java.util.Collection;
31 import java.util.List;
32 import java.util.Set;
33 import java.util.concurrent.ExecutionException;
34 import java.util.concurrent.ExecutorService;
35 import java.util.concurrent.Executors;
36 import java.util.concurrent.Future;
37 import java.util.concurrent.ThreadFactory;
38 import java.util.concurrent.atomic.AtomicLong;
39 import org.junit.After;
40 import org.junit.AfterClass;
41 import org.junit.Before;
42 import org.junit.BeforeClass;
43 import org.junit.Test;
44 import org.junit.runner.RunWith;
45 import org.junit.runners.Parameterized;
46 import org.opendaylight.netconf.api.DocumentedException;
47 import org.opendaylight.netconf.api.NetconfMessage;
48 import org.opendaylight.netconf.api.capability.Capability;
49 import org.opendaylight.netconf.api.messages.NetconfHelloMessageAdditionalHeader;
50 import org.opendaylight.netconf.api.xml.XmlNetconfConstants;
51 import org.opendaylight.netconf.api.xml.XmlUtil;
52 import org.opendaylight.netconf.client.NetconfClientDispatcher;
53 import org.opendaylight.netconf.client.NetconfClientDispatcherImpl;
54 import org.opendaylight.netconf.client.NetconfMessageUtil;
55 import org.opendaylight.netconf.client.SimpleNetconfClientSessionListener;
56 import org.opendaylight.netconf.client.TestingNetconfClient;
57 import org.opendaylight.netconf.client.conf.NetconfClientConfiguration;
58 import org.opendaylight.netconf.client.conf.NetconfClientConfigurationBuilder;
59 import org.opendaylight.netconf.nettyutil.NeverReconnectStrategy;
60 import org.opendaylight.netconf.nettyutil.handler.exi.NetconfStartExiMessage;
61 import org.opendaylight.netconf.server.api.SessionIdProvider;
62 import org.opendaylight.netconf.server.api.monitoring.CapabilityListener;
63 import org.opendaylight.netconf.server.api.monitoring.NetconfMonitoringService;
64 import org.opendaylight.netconf.server.api.monitoring.SessionEvent;
65 import org.opendaylight.netconf.server.api.monitoring.SessionListener;
66 import org.opendaylight.netconf.server.api.operations.HandlingPriority;
67 import org.opendaylight.netconf.server.api.operations.NetconfOperation;
68 import org.opendaylight.netconf.server.api.operations.NetconfOperationChainedExecution;
69 import org.opendaylight.netconf.server.api.operations.NetconfOperationService;
70 import org.opendaylight.netconf.server.api.operations.NetconfOperationServiceFactory;
71 import org.opendaylight.netconf.server.impl.DefaultSessionIdProvider;
72 import org.opendaylight.netconf.server.osgi.AggregatedNetconfOperationServiceFactory;
73 import org.opendaylight.netconf.test.util.XmlFileLoader;
74 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.base._1._0.rev110601.SessionIdType;
75 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.yang.ietf.netconf.monitoring.rev101004.netconf.state.CapabilitiesBuilder;
76 import org.opendaylight.yangtools.concepts.Registration;
77 import org.slf4j.Logger;
78 import org.slf4j.LoggerFactory;
79 import org.w3c.dom.Document;
80
81 @RunWith(Parameterized.class)
82 public class ConcurrentClientsTest {
83     private static final Logger LOG = LoggerFactory.getLogger(ConcurrentClientsTest.class);
84
85     private static ExecutorService clientExecutor;
86
87     private static final int CONCURRENCY = 32;
88     private static final InetSocketAddress NETCONF_ADDRESS = new InetSocketAddress("127.0.0.1", 8303);
89
90     private final int nettyThreads;
91     private final Class<? extends Runnable> clientRunnable;
92     private final Set<String> serverCaps;
93
94     public ConcurrentClientsTest(final int nettyThreads, final Class<? extends Runnable> clientRunnable,
95             final Set<String> serverCaps) {
96         this.nettyThreads = nettyThreads;
97         this.clientRunnable = clientRunnable;
98         this.serverCaps = serverCaps;
99     }
100
101     @Parameterized.Parameters()
102     public static Collection<Object[]> data() {
103         return List.of(new Object[][]{
104             { 4, TestingNetconfClientRunnable.class, NetconfServerSessionNegotiatorFactory.DEFAULT_BASE_CAPABILITIES},
105             { 1, TestingNetconfClientRunnable.class, NetconfServerSessionNegotiatorFactory.DEFAULT_BASE_CAPABILITIES},
106             // empty set of capabilities = only base 1.0 netconf capability
107             { 4, TestingNetconfClientRunnable.class, Set.of()},
108             { 4, TestingNetconfClientRunnable.class, getOnlyExiServerCaps()},
109             { 4, TestingNetconfClientRunnable.class, getOnlyChunkServerCaps()},
110             { 4, BlockingClientRunnable.class, getOnlyExiServerCaps()},
111             { 1, BlockingClientRunnable.class, getOnlyExiServerCaps()},
112         });
113     }
114
115     private EventLoopGroup nettyGroup;
116     private NetconfClientDispatcher netconfClientDispatcher;
117
118     HashedWheelTimer hashedWheelTimer;
119     private TestingNetconfOperation testingNetconfOperation;
120
121     public static NetconfMonitoringService createMockedMonitoringService() {
122         NetconfMonitoringService monitoring = mock(NetconfMonitoringService.class);
123         final SessionListener sessionListener = mock(SessionListener.class);
124         doNothing().when(sessionListener).onSessionUp(any(NetconfServerSession.class));
125         doNothing().when(sessionListener).onSessionDown(any(NetconfServerSession.class));
126         doNothing().when(sessionListener).onSessionEvent(any(SessionEvent.class));
127         doReturn((Registration) () -> { }).when(monitoring)
128             .registerCapabilitiesListener(any(NetconfMonitoringService.CapabilitiesListener.class));
129         doReturn(sessionListener).when(monitoring).getSessionListener();
130         doReturn(new CapabilitiesBuilder().setCapability(Set.of()).build()).when(monitoring).getCapabilities();
131         return monitoring;
132     }
133
134     @BeforeClass
135     public static void setUpClientExecutor() {
136         clientExecutor = Executors.newFixedThreadPool(CONCURRENCY, new ThreadFactory() {
137             int index = 1;
138
139             @Override
140             public Thread newThread(final Runnable runnable) {
141                 Thread thread = new Thread(runnable);
142                 thread.setName("client-" + index++);
143                 thread.setDaemon(true);
144                 return thread;
145             }
146         });
147     }
148
149     @Before
150     public void setUp() throws Exception {
151         hashedWheelTimer = new HashedWheelTimer();
152         nettyGroup = new NioEventLoopGroup(nettyThreads);
153         netconfClientDispatcher = new NetconfClientDispatcherImpl(nettyGroup, nettyGroup, hashedWheelTimer);
154
155         AggregatedNetconfOperationServiceFactory factoriesListener = new AggregatedNetconfOperationServiceFactory();
156
157         testingNetconfOperation = new TestingNetconfOperation();
158         factoriesListener.onAddNetconfOperationServiceFactory(
159                 new TestingOperationServiceFactory(testingNetconfOperation));
160
161         SessionIdProvider idProvider = new DefaultSessionIdProvider();
162
163         NetconfServerSessionNegotiatorFactory serverNegotiatorFactory = new
164                 NetconfServerSessionNegotiatorFactoryBuilder()
165                 .setTimer(hashedWheelTimer)
166                 .setAggregatedOpService(factoriesListener)
167                 .setIdProvider(idProvider)
168                 .setConnectionTimeoutMillis(5000)
169                 .setMonitoringService(createMockedMonitoringService())
170                 .setBaseCapabilities(serverCaps)
171                 .build();
172
173         ServerChannelInitializer serverChannelInitializer =
174                 new ServerChannelInitializer(serverNegotiatorFactory);
175         final NetconfServerDispatcherImpl dispatch =
176                 new NetconfServerDispatcherImpl(serverChannelInitializer, nettyGroup, nettyGroup);
177
178         ChannelFuture server = dispatch.createServer(NETCONF_ADDRESS);
179         server.await();
180     }
181
182     @After
183     public void tearDown() {
184         hashedWheelTimer.stop();
185         try {
186             nettyGroup.shutdownGracefully().get();
187         } catch (InterruptedException | ExecutionException e) {
188             LOG.warn("Ignoring exception while cleaning up after test", e);
189         }
190     }
191
192     @AfterClass
193     public static void tearDownClientExecutor() {
194         clientExecutor.shutdownNow();
195     }
196
197     @Test(timeout = CONCURRENCY * 1000)
198     public void testConcurrentClients() throws Exception {
199
200         List<Future<?>> futures = new ArrayList<>(CONCURRENCY);
201
202         for (int i = 0; i < CONCURRENCY; i++) {
203             futures.add(clientExecutor.submit(getInstanceOfClientRunnable()));
204         }
205
206         for (Future<?> future : futures) {
207             try {
208                 future.get();
209             } catch (InterruptedException e) {
210                 throw new IllegalStateException(e);
211             } catch (ExecutionException e) {
212                 LOG.error("Thread for testing client failed", e);
213                 throw e;
214             }
215         }
216
217         assertEquals(CONCURRENCY, testingNetconfOperation.getMessageCount());
218     }
219
220     public static Set<String> getOnlyExiServerCaps() {
221         return Set.of(
222                 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_0,
223                 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_CAPABILITY_EXI_1_0
224         );
225     }
226
227     public static Set<String> getOnlyChunkServerCaps() {
228         return Set.of(
229                 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_0,
230                 XmlNetconfConstants.URN_IETF_PARAMS_NETCONF_BASE_1_1
231         );
232     }
233
234     public Runnable getInstanceOfClientRunnable() throws Exception {
235         return clientRunnable.getConstructor(ConcurrentClientsTest.class).newInstance(this);
236     }
237
238     /**
239      * Responds to all operations except start-exi and counts all requests.
240      */
241     private static class TestingNetconfOperation implements NetconfOperation {
242
243         private final AtomicLong counter = new AtomicLong();
244
245         @Override
246         public HandlingPriority canHandle(final Document message) {
247             return XmlUtil.toString(message).contains(NetconfStartExiMessage.START_EXI)
248                     ? HandlingPriority.CANNOT_HANDLE :
249                     HandlingPriority.HANDLE_WITH_MAX_PRIORITY;
250         }
251
252         @SuppressWarnings("checkstyle:IllegalCatch")
253         @Override
254         public Document handle(final Document requestMessage,
255                 final NetconfOperationChainedExecution subsequentOperation) throws DocumentedException {
256             try {
257                 LOG.info("Handling netconf message from test {}", XmlUtil.toString(requestMessage));
258                 counter.getAndIncrement();
259                 return XmlUtil.readXmlToDocument("<test/>");
260             } catch (Exception e) {
261                 throw new RuntimeException(e);
262             }
263         }
264
265         public long getMessageCount() {
266             return counter.get();
267         }
268     }
269
270     /**
271      * Hardcoded operation service factory.
272      */
273     private static class TestingOperationServiceFactory implements NetconfOperationServiceFactory {
274         private final NetconfOperation[] operations;
275
276         TestingOperationServiceFactory(final NetconfOperation... operations) {
277             this.operations = operations;
278         }
279
280         @Override
281         public Set<Capability> getCapabilities() {
282             return Set.of();
283         }
284
285         @Override
286         public Registration registerCapabilityListener(final CapabilityListener listener) {
287             // No-op
288             return () -> { };
289         }
290
291         @Override
292         public NetconfOperationService createService(final SessionIdType sessionId) {
293             return new NetconfOperationService() {
294
295                 @Override
296                 public Set<NetconfOperation> getNetconfOperations() {
297                     return Set.of(operations);
298                 }
299
300                 @Override
301                 public void close() {
302                 }
303             };
304         }
305     }
306
307     /**
308      * Pure socket based blocking client.
309      */
310     public final class BlockingClientRunnable implements Runnable {
311
312         @Override
313         @SuppressWarnings("checkstyle:IllegalCatch")
314         public void run() {
315             try {
316                 run2();
317             } catch (Exception e) {
318                 throw new IllegalStateException(Thread.currentThread().getName(), e);
319             }
320         }
321
322         private void run2() throws Exception {
323             InputStream clientHello = requireNonNull(XmlFileLoader.getResourceAsStream(
324                 "netconfMessages/client_hello.xml"));
325             final InputStream getConfig = requireNonNull(XmlFileLoader.getResourceAsStream(
326                 "netconfMessages/getConfig.xml"));
327
328             Socket clientSocket = new Socket(NETCONF_ADDRESS.getHostString(), NETCONF_ADDRESS.getPort());
329             DataOutputStream outToServer = new DataOutputStream(clientSocket.getOutputStream());
330             InputStreamReader inFromServer = new InputStreamReader(clientSocket.getInputStream());
331
332             StringBuilder sb = new StringBuilder();
333             while (!sb.toString().endsWith("]]>]]>")) {
334                 sb.append((char) inFromServer.read());
335             }
336             LOG.info(sb.toString());
337
338             outToServer.write(ByteStreams.toByteArray(clientHello));
339             outToServer.write("]]>]]>".getBytes());
340             outToServer.flush();
341             // Thread.sleep(100);
342             outToServer.write(ByteStreams.toByteArray(getConfig));
343             outToServer.write("]]>]]>".getBytes());
344             outToServer.flush();
345             Thread.sleep(100);
346             sb = new StringBuilder();
347             while (!sb.toString().endsWith("]]>]]>")) {
348                 sb.append((char) inFromServer.read());
349             }
350             LOG.info(sb.toString());
351             clientSocket.close();
352         }
353     }
354
355     /**
356      * TestingNetconfClient based runnable.
357      */
358     public final class TestingNetconfClientRunnable implements Runnable {
359
360         @SuppressWarnings("checkstyle:IllegalCatch")
361         @Override
362         public void run() {
363             try {
364                 final TestingNetconfClient netconfClient =
365                         new TestingNetconfClient(Thread.currentThread().getName(), netconfClientDispatcher,
366                                 getClientConfig());
367                 final var sessionId = netconfClient.sessionId();
368                 LOG.info("Client with session id {}: hello exchanged", sessionId);
369
370                 final NetconfMessage getMessage = XmlFileLoader
371                         .xmlFileToNetconfMessage("netconfMessages/getConfig.xml");
372                 NetconfMessage result = netconfClient.sendRequest(getMessage).get();
373                 LOG.info("Client with session id {}: got result {}", sessionId.getValue(), result);
374
375                 checkState(NetconfMessageUtil.isErrorMessage(result) == false,
376                         "Received error response: " + XmlUtil.toString(result.getDocument()) + " to request: "
377                                 + XmlUtil.toString(getMessage.getDocument()));
378
379                 netconfClient.close();
380                 LOG.info("Client with session id {}: ended", sessionId.getValue());
381             } catch (final Exception e) {
382                 throw new IllegalStateException(Thread.currentThread().getName(), e);
383             }
384         }
385
386         private NetconfClientConfiguration getClientConfig() {
387             final NetconfClientConfigurationBuilder b = NetconfClientConfigurationBuilder.create();
388             b.withAddress(NETCONF_ADDRESS);
389             b.withAdditionalHeader(new NetconfHelloMessageAdditionalHeader("uname", "10.10.10.1", "830", "tcp",
390                     "client"));
391             b.withSessionListener(new SimpleNetconfClientSessionListener());
392             b.withReconnectStrategy(new NeverReconnectStrategy(GlobalEventExecutor.INSTANCE,
393                     NetconfClientConfigurationBuilder.DEFAULT_CONNECTION_TIMEOUT_MILLIS));
394             return b.build();
395         }
396     }
397 }