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

©2013 OpenDaylight, A Linux Foundation Collaborative Project. All Rights Reserved.
OpenDaylight is a registered trademark of The OpenDaylight Project, Inc.
Linux Foundation and OpenDaylight are registered trademarks of the Linux Foundation.
Linux is a registered trademark of Linus Torvalds.