netconf/tools/netconf-testtool/src/main/java/org/opendaylight/netconf/test/tool/client/stress/StressClient.java

   1 /*
   2  * Copyright (c) 2015 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.test.tool.client.stress;
   9
  10 import ch.qos.logback.classic.Level;
  11 import com.google.common.base.Stopwatch;
  12 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  13 import io.netty.channel.nio.NioEventLoopGroup;
  14 import io.netty.util.HashedWheelTimer;
  15 import io.netty.util.Timer;
  16 import java.io.IOException;
  17 import java.nio.file.Files;
  18 import java.util.ArrayList;
  19 import java.util.List;
  20 import java.util.concurrent.ExecutionException;
  21 import java.util.concurrent.Executors;
  22 import java.util.concurrent.TimeUnit;
  23 import java.util.concurrent.TimeoutException;
  24 import net.sourceforge.argparse4j.inf.ArgumentParserException;
  25 import org.opendaylight.netconf.api.messages.NetconfMessage;
  26 import org.opendaylight.netconf.api.xml.XmlUtil;
  27 import org.opendaylight.netconf.client.NetconfClientDispatcherImpl;
  28 import org.opendaylight.netconf.client.mdsal.NetconfDeviceCommunicator;
  29 import org.opendaylight.netconf.client.mdsal.api.NetconfSessionPreferences;
  30 import org.opendaylight.netconf.client.mdsal.api.RemoteDevice;
  31 import org.opendaylight.netconf.nettyutil.handler.ssh.client.AsyncSshHandler;
  32 import org.opendaylight.netconf.test.tool.TestToolUtils;
  33 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.base._1._0.rev110601.CommitInput;
  34 import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.base._1._0.rev110601.EditConfigInput;
  35 import org.opendaylight.yangtools.yang.common.QName;
  36 import org.slf4j.Logger;
  37 import org.slf4j.LoggerFactory;
  38 import org.w3c.dom.Document;
  39 import org.w3c.dom.Element;
  40 import org.xml.sax.SAXException;
  41
  42 public final class StressClient {
  43     private static final Logger LOG = LoggerFactory.getLogger(StressClient.class);
  44
  45     static final QName COMMIT_QNAME = QName.create(CommitInput.QNAME, "commit");
  46     public static final NetconfMessage COMMIT_MSG = new NetconfMessage(readString("""
  47         <rpc message-id="commit-batch" xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  48             <commit/>
  49         </rpc>"""));
  50
  51     static final QName EDIT_QNAME = QName.create(EditConfigInput.QNAME, "edit-config");
  52     static final Document EDIT_CANDIDATE_BLUEPRINT = readString("""
  53         <rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  54             <edit-config>
  55                 <target>
  56                     <candidate/>
  57                 </target>
  58                 <default-operation>none</default-operation>
  59                 <config/>
  60             </edit-config>
  61         </rpc>""");
  62     static final Document EDIT_RUNNING_BLUEPRINT  = readString("""
  63         <rpc xmlns="urn:ietf:params:xml:ns:netconf:base:1.0">
  64             <edit-config>
  65                 <target>
  66                     <running/>
  67                 </target>
  68                 <default-operation>none</default-operation>
  69                 <config/>
  70             </edit-config>
  71         </rpc>""");
  72
  73     private static Document readString(final String str) {
  74         try {
  75             return XmlUtil.readXmlToDocument(str);
  76         } catch (SAXException | IOException e) {
  77             throw new ExceptionInInitializerError(e);
  78         }
  79     }
  80
  81     private static final String MSG_ID_PLACEHOLDER_REGEX = "\\{MSG_ID\\}";
  82     private static final String PHYS_ADDR_PLACEHOLDER = "{PHYS_ADDR}";
  83
  84     private static long macStart = 0xAABBCCDD0000L;
  85
  86     private static Parameters params;
  87
  88     private StressClient() {
  89         // Hidden on purpose
  90     }
  91
  92     public static void main(final String[] args) throws ExecutionException, InterruptedException, TimeoutException {
  93         if (initParameters(args)) {
  94             return;
  95         }
  96         params.validate();
  97
  98         final var root = (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);
  99         root.setLevel(params.debug ? Level.DEBUG : Level.INFO);
 100
 101         final int threadAmount = params.threadAmount;
 102         LOG.info("thread amount: {}", threadAmount);
 103         final int requestsPerThread = params.editCount / params.threadAmount;
 104         LOG.info("requestsPerThread: {}", requestsPerThread);
 105         final int leftoverRequests = params.editCount % params.threadAmount;
 106         LOG.info("leftoverRequests: {}", leftoverRequests);
 107
 108         LOG.info("Preparing messages");
 109         // Prepare all msgs up front
 110         final var allPreparedMessages = new ArrayList<List<NetconfMessage>>(threadAmount);
 111         for (int i = 0; i < threadAmount; i++) {
 112             if (i != threadAmount - 1) {
 113                 allPreparedMessages.add(new ArrayList<>(requestsPerThread));
 114             } else {
 115                 allPreparedMessages.add(new ArrayList<>(requestsPerThread + leftoverRequests));
 116             }
 117         }
 118
 119
 120         final String editContentString;
 121         try {
 122             editContentString = Files.readString(params.editContent.toPath());
 123         } catch (final IOException e) {
 124             throw new IllegalArgumentException("Cannot read content of " + params.editContent, e);
 125         }
 126
 127         for (int i = 0; i < threadAmount; i++) {
 128             final var preparedMessages = allPreparedMessages.get(i);
 129             int padding = 0;
 130             if (i == threadAmount - 1) {
 131                 padding = leftoverRequests;
 132             }
 133             for (int j = 0; j < requestsPerThread + padding; j++) {
 134                 LOG.debug("id: {}", i * requestsPerThread + j);
 135                 preparedMessages.add(prepareMessage(i * requestsPerThread + j, editContentString));
 136             }
 137         }
 138
 139         final var nioGroup = new NioEventLoopGroup();
 140         final var timer = new HashedWheelTimer();
 141
 142         final var netconfClientDispatcher = configureClientDispatcher(nioGroup, timer);
 143
 144         final var callables = new ArrayList<StressClientCallable>(threadAmount);
 145         for (var messages : allPreparedMessages) {
 146             callables.add(new StressClientCallable(params, netconfClientDispatcher, messages));
 147         }
 148
 149         final var executorService = Executors.newFixedThreadPool(threadAmount);
 150
 151         LOG.info("Starting stress test");
 152         final var sw = Stopwatch.createStarted();
 153         final var futures = executorService.invokeAll(callables);
 154         for (var future : futures) {
 155             future.get(4L, TimeUnit.MINUTES);
 156         }
 157         executorService.shutdownNow();
 158         sw.stop();
 159
 160         LOG.info("FINISHED. Execution time: {}", sw);
 161         LOG.info("Requests per second: {}", params.editCount * 1000.0 / sw.elapsed(TimeUnit.MILLISECONDS));
 162
 163         // Cleanup
 164         timer.stop();
 165         try {
 166             nioGroup.shutdownGracefully().get(20L, TimeUnit.SECONDS);
 167         } catch (InterruptedException | ExecutionException | TimeoutException e) {
 168             LOG.warn("Unable to close executor properly", e);
 169         }
 170         //stop the underlying ssh thread that gets spawned if we use ssh
 171         if (params.ssh) {
 172             AsyncSshHandler.DEFAULT_CLIENT.stop();
 173         }
 174     }
 175
 176     static NetconfMessage prepareMessage(final int id, final String editContentString) {
 177         final Document msg = XmlUtil.createDocumentCopy(
 178             params.candidateDatastore ? EDIT_CANDIDATE_BLUEPRINT : EDIT_RUNNING_BLUEPRINT);
 179         msg.getDocumentElement().setAttribute("message-id", Integer.toString(id));
 180         final NetconfMessage netconfMessage = new NetconfMessage(msg);
 181
 182         final Element editContentElement;
 183         try {
 184             // Insert message id where needed
 185             String specificEditContent = editContentString.replaceAll(MSG_ID_PLACEHOLDER_REGEX, Integer.toString(id));
 186
 187             final var sb = new StringBuilder(specificEditContent);
 188             int idx = sb.indexOf(PHYS_ADDR_PLACEHOLDER);
 189             while (idx != -1) {
 190                 sb.replace(idx, idx + PHYS_ADDR_PLACEHOLDER.length(), TestToolUtils.getMac(macStart++));
 191                 idx = sb.indexOf(PHYS_ADDR_PLACEHOLDER);
 192             }
 193             specificEditContent = sb.toString();
 194
 195             editContentElement = XmlUtil.readXmlToElement(specificEditContent);
 196             final var config = ((Element) msg.getDocumentElement().getElementsByTagName("edit-config").item(0))
 197                     .getElementsByTagName("config").item(0);
 198             config.appendChild(msg.importNode(editContentElement, true));
 199         } catch (final IOException | SAXException e) {
 200             throw new IllegalArgumentException("Edit content file is unreadable", e);
 201         }
 202
 203         return netconfMessage;
 204     }
 205
 206     @SuppressFBWarnings(value = "DM_EXIT", justification = "Exit from CLI with error without throwing an exception")
 207     private static boolean initParameters(final String[] args) {
 208         final var parser = Parameters.getParser();
 209         params = new Parameters();
 210         try {
 211             parser.parseArgs(args, args);
 212         } catch (ArgumentParserException e) {
 213             parser.handleError(e);
 214             System.exit(1);
 215             return true;
 216         }
 217         return false;
 218     }
 219
 220     @Deprecated
 221     private static NetconfClientDispatcherImpl configureClientDispatcher(final NioEventLoopGroup nioGroup,
 222             final Timer timer) {
 223         if (params.exi) {
 224             return params.legacyFraming ? ConfigurableClientDispatcher.createLegacyExi(nioGroup, nioGroup, timer)
 225                 : ConfigurableClientDispatcher.createChunkedExi(nioGroup, nioGroup, timer);
 226         }
 227         return params.legacyFraming ? ConfigurableClientDispatcher.createLegacy(nioGroup, nioGroup, timer)
 228             : ConfigurableClientDispatcher.createChunked(nioGroup, nioGroup, timer);
 229     }
 230
 231     static class LoggingRemoteDevice implements RemoteDevice<NetconfDeviceCommunicator> {
 232         @Override
 233         public void onRemoteSessionUp(final NetconfSessionPreferences remoteSessionCapabilities,
 234                                       final NetconfDeviceCommunicator netconfDeviceCommunicator) {
 235             LOG.info("Session established");
 236         }
 237
 238         @Override
 239         public void onRemoteSessionDown() {
 240             LOG.info("Session down");
 241         }
 242
 243         @Override
 244         public void onNotification(final NetconfMessage notification) {
 245             LOG.info("Notification received: {}", notification);
 246         }
 247     }
 248 }