<feature name='odl-netconf-mdsal' version='${project.version}' description="OpenDaylight :: Netconf :: Mdsal">
<feature version='${config.version}'>odl-config-all</feature>
<feature version='${netconf.version}'>odl-netconf-all</feature>
+ <feature version='${config.version}'>odl-config-netty</feature>
+ <bundle>mvn:org.opendaylight.controller/netconf-tcp/${netconf.version}</bundle>
<bundle>mvn:org.opendaylight.controller/netconf-ssh/${netconf.version}</bundle>
<feature version='${mdsal.version}'>odl-mdsal-broker</feature>
<bundle>mvn:org.opendaylight.controller/mdsal-netconf-connector/${netconf.version}</bundle>
<!-- feature version='${aaa.version}'>odl-aaa-netconf-plugin</feature -->
<bundle>mvn:org.opendaylight.controller/netconf-ssh/${netconf.version}</bundle>
</feature>
- <feature name='odl-netconf-tcp' version='${netconf.version}' description="OpenDaylight :: Netconf Connector :: TCP">
+ <feature name='odl-netconf-tcp' version='${netconf.version}' description="OpenDaylight :: Netconf Connector :: TCP">
<feature version='${netconf.version}'>odl-netconf-impl</feature>
+ <feature version='${config.version}'>odl-config-netty</feature>
<bundle>mvn:org.opendaylight.controller/netconf-tcp/${netconf.version}</bundle>
- </feature>
+ </feature>
<feature name='odl-message-bus' version='${project.version}'>
<feature version='${project.version}'>odl-netconf-connector</feature>
<properties>
- <akka.version>2.3.9</akka.version>
+ <akka.version>2.3.10</akka.version>
<appauth.version>0.6.0-SNAPSHOT</appauth.version>
<archetype-app-northbound>0.2.0-SNAPSHOT</archetype-app-northbound>
<arphandler.version>0.7.0-SNAPSHOT</arphandler.version>
.child(TopologyTypes.class)
.augmentation(TopologyTypes1.class);
- private final Map<DataChangeListener, ListenerRegistration<DataChangeListener>> topicListenerRegistrations =
+ private final Map<EventSourceTopic, ListenerRegistration<DataChangeListener>> topicListenerRegistrations =
new ConcurrentHashMap<>();
private final Map<NodeKey, RoutedRpcRegistration<EventSourceService>> routedRpcRegistrations =
- new ConcurrentHashMap<>();;
+ new ConcurrentHashMap<>();
private final DataBroker dataBroker;
private final RpcRegistration<EventAggregatorService> aggregatorRpcReg;
}
public void register(final EventSource eventSource){
- NodeKey nodeKey = eventSource.getSourceNodeKey();
+ NodeKey nodeKey = eventSource.getSourceNodeKey();
final KeyedInstanceIdentifier<Node, NodeKey> sourcePath = EVENT_SOURCE_TOPOLOGY_PATH.child(Node.class, nodeKey);
RoutedRpcRegistration<EventSourceService> reg = rpcRegistry.addRoutedRpcImplementation(EventSourceService.class, eventSource);
reg.registerPath(NodeContext.class, sourcePath);
routedRpcRegistrations.put(nodeKey,reg);
insert(sourcePath);
+
+ for(EventSourceTopic est : topicListenerRegistrations.keySet()){
+ est.notifyNode(EVENT_SOURCE_TOPOLOGY_PATH.child(Node.class, nodeKey));
+ }
}
public void unRegister(final EventSource eventSource){
result = partialResult;
}
return new NormalizedNodeContext(path,result);
+ } catch (final RestconfDocumentedException e) {
+ throw e;
} catch (final Exception e) {
LOG.debug("Error parsing json input", e);
final NormalizedNode<?, ?> result = parse(path,doc);
return new NormalizedNodeContext(path,result);
+ } catch (final RestconfDocumentedException e){
+ throw e;
} catch (final Exception e) {
LOG.debug("Error parsing xml input", e);
final InstanceIdentifierBuilder builder = YangInstanceIdentifier.builder();
final Module latestModule = globalSchema.findModuleByName(startModule, null);
+
+ if (latestModule == null) {
+ throw new RestconfDocumentedException("The module named '" + startModule + "' does not exist.", ErrorType.PROTOCOL, ErrorTag.UNKNOWN_ELEMENT);
+ }
+
final InstanceIdentifierContext<?> iiWithSchemaNode = collectPathArguments(builder, pathArgs, latestModule, null,
toMountPointIdentifier);
return retValue;
}
throw new RestconfDocumentedException("RpcError message", null, retValue.getErrors());
- }
- catch (final InterruptedException e) {
+ } catch (final InterruptedException e) {
throw new RestconfDocumentedException(
"The operation was interrupted while executing and did not complete.", ErrorType.RPC,
ErrorTag.PARTIAL_OPERATION);
- }
- catch (final ExecutionException e) {
+ } catch (final ExecutionException e) {
Throwable cause = e.getCause();
- if (cause instanceof CancellationException) {
- throw new RestconfDocumentedException("The operation was cancelled while executing.", ErrorType.RPC,
- ErrorTag.PARTIAL_OPERATION);
- } else if (cause != null) {
+ if (cause != null) {
while (cause.getCause() != null) {
cause = cause.getCause();
}
throw new RestconfDocumentedException("The operation encountered an unexpected error while executing.",
e);
}
+ } catch (final CancellationException e) {
+ throw new RestconfDocumentedException("The operation was cancelled while executing.", ErrorType.RPC,
+ ErrorTag.PARTIAL_OPERATION);
}
}
import java.util.Collection;
import java.util.Collections;
import org.opendaylight.controller.config.yang.md.sal.rest.connector.Config;
+import org.opendaylight.controller.config.yang.md.sal.rest.connector.Delete;
import org.opendaylight.controller.config.yang.md.sal.rest.connector.Get;
import org.opendaylight.controller.config.yang.md.sal.rest.connector.Operational;
import org.opendaylight.controller.config.yang.md.sal.rest.connector.Post;
@Override
public Config getConfig() {
final Config config = new Config();
+
final Get get = new Get();
get.setReceivedRequests(stats.getConfigGet());
+ get.setSuccessfulResponses(stats.getSuccessGetConfig());
+ get.setFailedResponses(stats.getFailureGetConfig());
config.setGet(get);
+
final Post post = new Post();
post.setReceivedRequests(stats.getConfigPost());
+ post.setSuccessfulResponses(stats.getSuccessPost());
+ post.setFailedResponses(stats.getFailurePost());
config.setPost(post);
+
final Put put = new Put();
put.setReceivedRequests(stats.getConfigPut());
+ put.setSuccessfulResponses(stats.getSuccessPut());
+ put.setFailedResponses(stats.getFailurePut());
config.setPut(put);
+
+ final Delete delete = new Delete();
+ delete.setReceivedRequests(stats.getConfigDelete());
+ delete.setSuccessfulResponses(stats.getSuccessDelete());
+ delete.setFailedResponses(stats.getFailureDelete());
+ config.setDelete(delete);
+
return config;
}
final Operational operational = new Operational();
final Get get = new Get();
get.setReceivedRequests(opGet);
+ get.setSuccessfulResponses(stats.getSuccessGetOperational());
+ get.setFailedResponses(stats.getFailureGetOperational());
operational.setGet(get);
return operational;
}
final BigInteger rpcInvoke = stats.getRpc();
final Rpcs rpcs = new Rpcs();
rpcs.setReceivedRequests(rpcInvoke);
- return rpcs ;
+ return rpcs;
}
-}
+}
\ No newline at end of file
import java.math.BigInteger;
import java.util.concurrent.atomic.AtomicLong;
import javax.ws.rs.core.Response;
+import javax.ws.rs.core.Response.Status;
import javax.ws.rs.core.UriInfo;
import org.opendaylight.controller.sal.rest.api.RestconfService;
AtomicLong configPost = new AtomicLong();
AtomicLong configPut = new AtomicLong();
AtomicLong configDelete = new AtomicLong();
+ AtomicLong successGetConfig = new AtomicLong();
+ AtomicLong successGetOperational = new AtomicLong();
+ AtomicLong successPost = new AtomicLong();
+ AtomicLong successPut = new AtomicLong();
+ AtomicLong successDelete = new AtomicLong();
+ AtomicLong failureGetConfig = new AtomicLong();
+ AtomicLong failureGetOperational = new AtomicLong();
+ AtomicLong failurePost = new AtomicLong();
+ AtomicLong failurePut = new AtomicLong();
+ AtomicLong failureDelete = new AtomicLong();
private static final StatisticsRestconfServiceWrapper INSTANCE = new StatisticsRestconfServiceWrapper(RestconfImpl.getInstance());
@Override
public NormalizedNodeContext readConfigurationData(final String identifier, final UriInfo uriInfo) {
configGet.incrementAndGet();
- return delegate.readConfigurationData(identifier, uriInfo);
+ NormalizedNodeContext normalizedNodeContext = null;
+ try {
+ normalizedNodeContext = delegate.readConfigurationData(identifier, uriInfo);
+ if (normalizedNodeContext.getData() != null) {
+ successGetConfig.incrementAndGet();
+ }
+ else {
+ failureGetConfig.incrementAndGet();
+ }
+ } catch (Exception e) {
+ failureGetConfig.incrementAndGet();
+ throw e;
+ }
+ return normalizedNodeContext;
}
@Override
public NormalizedNodeContext readOperationalData(final String identifier, final UriInfo uriInfo) {
operationalGet.incrementAndGet();
- return delegate.readOperationalData(identifier, uriInfo);
+ NormalizedNodeContext normalizedNodeContext = null;
+ try {
+ normalizedNodeContext = delegate.readOperationalData(identifier, uriInfo);
+ if (normalizedNodeContext.getData() != null) {
+ successGetOperational.incrementAndGet();
+ }
+ else {
+ failureGetOperational.incrementAndGet();
+ }
+ } catch (Exception e) {
+ failureGetOperational.incrementAndGet();
+ throw e;
+ }
+ return normalizedNodeContext;
}
@Override
public Response updateConfigurationData(final String identifier, final NormalizedNodeContext payload) {
configPut.incrementAndGet();
- return delegate.updateConfigurationData(identifier, payload);
+ Response response = null;
+ try {
+ response = delegate.updateConfigurationData(identifier, payload);
+ if (response.getStatus() == Status.OK.getStatusCode()) {
+ successPut.incrementAndGet();
+ }
+ else {
+ failurePut.incrementAndGet();
+ }
+ } catch (Exception e) {
+ failurePut.incrementAndGet();
+ throw e;
+ }
+ return response;
}
@Override
public Response createConfigurationData(final String identifier, final NormalizedNodeContext payload, final UriInfo uriInfo) {
configPost.incrementAndGet();
- return delegate.createConfigurationData(identifier, payload, uriInfo);
+ Response response = null;
+ try {
+ response = delegate.createConfigurationData(identifier, payload, uriInfo);
+ if (response.getStatus() == Status.OK.getStatusCode()) {
+ successPost.incrementAndGet();
+ }
+ else {
+ failurePost.incrementAndGet();
+ }
+ } catch (Exception e) {
+ failurePost.incrementAndGet();
+ throw e;
+ }
+ return response;
}
@Override
public Response createConfigurationData(final NormalizedNodeContext payload, final UriInfo uriInfo) {
configPost.incrementAndGet();
- return delegate.createConfigurationData(payload, uriInfo);
+ Response response = null;
+ try {
+ response = delegate.createConfigurationData(payload, uriInfo);
+ if (response.getStatus() == Status.OK.getStatusCode()) {
+ successPost.incrementAndGet();
+ }
+ else {
+ failurePost.incrementAndGet();
+ }
+ }catch (Exception e) {
+ failurePost.incrementAndGet();
+ throw e;
+ }
+ return response;
}
@Override
public Response deleteConfigurationData(final String identifier) {
- return delegate.deleteConfigurationData(identifier);
+ configDelete.incrementAndGet();
+ Response response = null;
+ try {
+ response = delegate.deleteConfigurationData(identifier);
+ if (response.getStatus() == Status.OK.getStatusCode()) {
+ successDelete.incrementAndGet();
+ }
+ else {
+ failureDelete.incrementAndGet();
+ }
+ } catch (Exception e) {
+ failureDelete.incrementAndGet();
+ throw e;
+ }
+ return response;
}
@Override
public BigInteger getRpc() {
return BigInteger.valueOf(rpc.get());
}
-}
+
+ public BigInteger getSuccessGetConfig() {
+ return BigInteger.valueOf(successGetConfig.get());
+ }
+
+ public BigInteger getSuccessGetOperational() {
+ return BigInteger.valueOf(successGetOperational.get());
+ }
+
+ public BigInteger getSuccessPost() {
+ return BigInteger.valueOf(successPost.get());
+ }
+
+ public BigInteger getSuccessPut() {
+ return BigInteger.valueOf(successPut.get());
+ }
+
+ public BigInteger getSuccessDelete() {
+ return BigInteger.valueOf(successDelete.get());
+ }
+
+ public BigInteger getFailureGetConfig() {
+ return BigInteger.valueOf(failureGetConfig.get());
+ }
+
+ public BigInteger getFailureGetOperational() {
+ return BigInteger.valueOf(failureGetOperational.get());
+ }
+
+ public BigInteger getFailurePost() {
+ return BigInteger.valueOf(failurePost.get());
+ }
+
+ public BigInteger getFailurePut() {
+ return BigInteger.valueOf(failurePut.get());
+ }
+
+ public BigInteger getFailureDelete() {
+ return BigInteger.valueOf(failureDelete.get());
+ }
+}
\ No newline at end of file
leaf received-requests {
type uint64;
}
+
+ leaf successful-responses {
+ type uint64;
+ }
+
+ leaf failed-responses {
+ type uint64;
+ }
}
augment "/config:modules/config:module/config:configuration" {
container put {
uses statistics;
}
+
+ container delete {
+ uses statistics;
+ }
}
container operational {
tx.put(LogicalDatastoreType.OPERATIONAL, InstanceIdentifier.create(NetconfState.class), state);
// FIXME first attempt (right after we register to binding broker) always fails
// Is it due to the fact that we are writing from the onSessionInitiated callback ?
- final CheckedFuture<Void, TransactionCommitFailedException> submit = tx.submit();
-
- Futures.addCallback(submit, new FutureCallback<Void>() {
- @Override
- public void onSuccess(final Void aVoid) {
- LOG.debug("Netconf state updated successfully");
- }
-
- @Override
- public void onFailure(final Throwable throwable) {
- LOG.warn("Unable to update netconf state", throwable);
- }
- });
+ try {
+ tx.submit().checkedGet();
+ LOG.debug("Netconf state updated successfully");
+ } catch (TransactionCommitFailedException e) {
+ LOG.warn("Unable to update netconf state", e);
+ }
}
@Override
<password>admin</password>
</module>
+
+ <!--TCP endpoint for MD-SAL netconf server -->
+ <!--<module>-->
+ <!--<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:tcp">prefix:netconf-northbound-tcp</type>-->
+ <!--<name>netconf-mdsal-tcp-server</name>-->
+ <!--<dispatcher xmlns="urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:tcp">-->
+ <!--<type xmlns:prefix="urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound">prefix:netconf-server-dispatcher</type>-->
+ <!--<name>netconf-mdsal-server-dispatcher</name>-->
+ <!--</dispatcher>-->
+ <!--</module>-->
+
</modules>
<services xmlns="urn:opendaylight:params:xml:ns:yang:controller:config">
<capability>urn:opendaylight:params:xml:ns:yang:controller:netconf:mdsal:mapper?module=netconf-mdsal-mapper&revision=2015-01-14</capability>
<capability>urn:opendaylight:params:xml:ns:yang:controller:netconf:mdsal:monitoring?module=netconf-mdsal-monitoring&revision=2015-02-18</capability>
<capability>urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:ssh?module=netconf-northbound-ssh&revision=2015-01-14</capability>
+ <capability>urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:tcp?module=netconf-northbound-tcp&revision=2015-04-23</capability>
<capability>urn:opendaylight:params:xml:ns:yang:controller:config:netconf:northbound:impl?module=netconf-northbound-impl&revision=2015-01-12</capability>
<capability>urn:opendaylight:params:xml:ns:yang:controller:threadpool:impl:scheduled?module=threadpool-impl-scheduled&revision=2013-12-01</capability>
</required-capabilities>
<groupId>${project.groupId}</groupId>
<artifactId>netconf-util</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>threadpool-config-api</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>netty-config-api</artifactId>
+ </dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</instructions>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.opendaylight.yangtools</groupId>
+ <artifactId>yang-maven-plugin</artifactId>
+ <executions>
+ <execution>
+ <id>config</id>
+ <goals>
+ <goal>generate-sources</goal>
+ </goals>
+ <configuration>
+ <codeGenerators>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator</codeGeneratorClass>
+ <outputBaseDir>${jmxGeneratorPath}</outputBaseDir>
+ <additionalConfiguration>
+ <namespaceToPackage1>urn:opendaylight:params:xml:ns:yang:controller==org.opendaylight.controller.config.yang</namespaceToPackage1>
+ </additionalConfiguration>
+ </generator>
+ <generator>
+ <codeGeneratorClass>org.opendaylight.yangtools.maven.sal.api.gen.plugin.CodeGeneratorImpl</codeGeneratorClass>
+ <outputBaseDir>${salGeneratorPath}</outputBaseDir>
+ </generator>
+ </codeGenerators>
+ <inspectDependencies>true</inspectDependencies>
+ </configuration>
+ </execution>
+ </executions>
+ <dependencies>
+ <dependency>
+ <groupId>org.opendaylight.controller</groupId>
+ <artifactId>yang-jmx-generator-plugin</artifactId>
+ <version>${config.version}</version>
+ </dependency>
+ </dependencies>
+ </plugin>
</plugins>
</build>
--- /dev/null
+package org.opendaylight.controller.config.yang.netconf.northbound.tcp;
+
+import io.netty.channel.ChannelFuture;
+import io.netty.util.concurrent.GenericFutureListener;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.net.UnknownHostException;
+import org.opendaylight.controller.netconf.api.NetconfServerDispatcher;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class NetconfNorthboundTcpModule extends org.opendaylight.controller.config.yang.netconf.northbound.tcp.AbstractNetconfNorthboundTcpModule {
+
+ private static final Logger LOG = LoggerFactory.getLogger(NetconfNorthboundTcpModule.class);
+
+
+ public NetconfNorthboundTcpModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver) {
+ super(identifier, dependencyResolver);
+ }
+
+ public NetconfNorthboundTcpModule(org.opendaylight.controller.config.api.ModuleIdentifier identifier, org.opendaylight.controller.config.api.DependencyResolver dependencyResolver, org.opendaylight.controller.config.yang.netconf.northbound.tcp.NetconfNorthboundTcpModule oldModule, java.lang.AutoCloseable oldInstance) {
+ super(identifier, dependencyResolver, oldModule, oldInstance);
+ }
+
+ @Override
+ public void customValidation() {
+ // add custom validation form module attributes here.
+ }
+
+ @Override
+ public java.lang.AutoCloseable createInstance() {
+ final NetconfServerDispatcher dispatch = getDispatcherDependency();
+ final ChannelFuture tcpServer = dispatch.createServer(getInetAddress());
+
+ tcpServer.addListener(new GenericFutureListener<ChannelFuture>() {
+ @Override
+ public void operationComplete(ChannelFuture future) throws Exception {
+ if (future.isDone() && future.isSuccess()) {
+ LOG.info("Netconf TCP endpoint started successfully at {}", getInetAddress());
+ } else {
+ LOG.warn("Unable to start TCP netconf server at {}", getInetAddress(), future.cause());
+ throw new RuntimeException("Unable to start TCP netconf server", future.cause());
+ }
+ }
+ });
+
+ return new NetconfServerCloseable(tcpServer);
+ }
+
+ private InetSocketAddress getInetAddress() {
+ try {
+ final InetAddress inetAd = InetAddress.getByName(getBindingAddress().getIpv4Address() == null ? getBindingAddress().getIpv6Address().getValue() : getBindingAddress().getIpv4Address().getValue());
+ return new InetSocketAddress(inetAd, getPort().getValue());
+ } catch (final UnknownHostException e) {
+ throw new IllegalArgumentException("Unable to bind netconf endpoint to address " + getBindingAddress(), e);
+ }
+ }
+
+ private static final class NetconfServerCloseable implements AutoCloseable {
+ private final ChannelFuture localServer;
+
+ public NetconfServerCloseable(final ChannelFuture localServer) {
+ this.localServer = localServer;
+ }
+
+ @Override
+ public void close() throws Exception {
+ if(localServer.isDone()) {
+ localServer.channel().close();
+ } else {
+ localServer.cancel(true);
+ }
+ }
+ }
+
+}
--- /dev/null
+/*
+* Generated file
+*
+* Generated from: yang module name: netconf-northbound-tcp yang module local name: netconf-northbound-tcp
+* Generated by: org.opendaylight.controller.config.yangjmxgenerator.plugin.JMXGenerator
+* Generated at: Thu Apr 23 16:34:55 CEST 2015
+*
+* Do not modify this file unless it is present under src/main directory
+*/
+package org.opendaylight.controller.config.yang.netconf.northbound.tcp;
+public class NetconfNorthboundTcpModuleFactory extends org.opendaylight.controller.config.yang.netconf.northbound.tcp.AbstractNetconfNorthboundTcpModuleFactory {
+
+}
--- /dev/null
+module netconf-northbound-tcp {
+ yang-version 1;
+ namespace "urn:opendaylight:params:xml:ns:yang:controller:netconf:northbound:tcp";
+ prefix "nni";
+
+ import netconf-northbound-mapper { prefix nnm; revision-date 2015-01-14; }
+ import netconf-northbound { prefix nn; revision-date 2015-01-14; }
+ import config { prefix config; revision-date 2013-04-05; }
+ import threadpool {prefix th;}
+ import netty {prefix netty;}
+ import ietf-inet-types { prefix inet; revision-date 2010-09-24; }
+
+ organization "Cisco Systems, Inc.";
+
+ description
+ "This module contains the base YANG definitions for
+ a default implementation of netconf northbound tcp server";
+
+ revision "2015-04-23" {
+ description
+ "Initial revision.";
+ }
+
+ identity netconf-northbound-tcp {
+ base config:module-type;
+ config:java-name-prefix NetconfNorthboundTcp;
+ }
+
+ augment "/config:modules/config:module/config:configuration" {
+ case netconf-northbound-tcp {
+ when "/config:modules/config:module/config:type = 'netconf-northbound-tcp'";
+
+ leaf port {
+ type inet:port-number;
+ default 2831;
+ }
+
+ leaf binding-address {
+ type inet:ip-address;
+ default "0.0.0.0";
+ }
+
+ container dispatcher {
+ uses config:service-ref {
+ refine type {
+ config:required-identity nn:netconf-server-dispatcher;
+ }
+ }
+ }
+ }
+ }
+
+}
\ No newline at end of file
private final List<NetconfMessage> preparedMessages;
private final NetconfDeviceCommunicator sessionListener;
private final List<Integer> editBatches;
+ private final int editAmount;
public AsyncExecutionStrategy(final Parameters params, final List<NetconfMessage> editConfigMsgs, final NetconfDeviceCommunicator sessionListener) {
this.params = params;
this.preparedMessages = editConfigMsgs;
this.sessionListener = sessionListener;
- this.editBatches = countEditBatchSizes(params);
+ this.editBatches = countEditBatchSizes(params, editConfigMsgs.size());
+ editAmount = editConfigMsgs.size();
}
- private static List<Integer> countEditBatchSizes(final Parameters params) {
+ private static List<Integer> countEditBatchSizes(final Parameters params, final int amount) {
final List<Integer> editBatches = Lists.newArrayList();
- if (params.editBatchSize != params.editCount) {
- final int fullBatches = params.editCount / params.editBatchSize;
+ if (params.editBatchSize != amount) {
+ final int fullBatches = amount / params.editBatchSize;
for (int i = 0; i < fullBatches; i++) {
editBatches.add(params.editBatchSize);
}
- if (params.editCount % params.editBatchSize != 0) {
- editBatches.add(params.editCount % params.editBatchSize);
+ if (amount % params.editBatchSize != 0) {
+ editBatches.add(amount % params.editBatchSize);
}
} else {
editBatches.add(params.editBatchSize);
}
}
- Preconditions.checkState(responseCounter.get() == params.editCount + editBatches.size(), "Not all responses were received, only %s from %s", responseCounter.get(), params.editCount + editBatches.size());
+ Preconditions.checkState(responseCounter.get() == editAmount + editBatches.size(), "Not all responses were received, only %s from %s", responseCounter.get(), params.editCount + editBatches.size());
}
}
@Arg(dest = "tcp-header")
public String tcpHeader;
+ @Arg(dest = "thread-amount")
+ public int threadAmount;
+
static ArgumentParser getParser() {
final ArgumentParser parser = ArgumentParsers.newArgumentParser("netconf stress client");
.required(false)
.dest("tcp-header");
+ parser.addArgument("--thread-amount")
+ .type(Integer.class)
+ .setDefault(1)
+ .dest("thread-amount");
+
// TODO add get-config option instead of edit + commit
// TODO different edit config content
import ch.qos.logback.classic.Level;
import com.google.common.base.Charsets;
-import com.google.common.base.Function;
-import com.google.common.base.Joiner;
-import com.google.common.base.Splitter;
import com.google.common.base.Stopwatch;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
import com.google.common.io.Files;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.HashedWheelTimer;
import io.netty.util.Timer;
-import io.netty.util.concurrent.GlobalEventExecutor;
import java.io.IOException;
-import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import net.sourceforge.argparse4j.inf.ArgumentParser;
import net.sourceforge.argparse4j.inf.ArgumentParserException;
import org.opendaylight.controller.netconf.api.NetconfMessage;
import org.opendaylight.controller.netconf.client.NetconfClientDispatcherImpl;
-import org.opendaylight.controller.netconf.client.NetconfClientSession;
-import org.opendaylight.controller.netconf.client.conf.NetconfClientConfiguration;
-import org.opendaylight.controller.netconf.client.conf.NetconfClientConfigurationBuilder;
-import org.opendaylight.controller.netconf.util.messages.NetconfHelloMessageAdditionalHeader;
import org.opendaylight.controller.netconf.util.xml.XmlUtil;
import org.opendaylight.controller.sal.connect.api.RemoteDevice;
import org.opendaylight.controller.sal.connect.netconf.listener.NetconfDeviceCommunicator;
import org.opendaylight.controller.sal.connect.netconf.listener.NetconfSessionPreferences;
-import org.opendaylight.controller.sal.connect.util.RemoteDeviceId;
-import org.opendaylight.protocol.framework.NeverReconnectStrategy;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.base._1._0.rev110601.CommitInput;
import org.opendaylight.yang.gen.v1.urn.ietf.params.xml.ns.netconf.base._1._0.rev110601.EditConfigInput;
import org.opendaylight.yangtools.yang.common.QName;
" <target>\n" +
" <candidate/>\n" +
" </target>\n" +
+ " <default-operation>none</default-operation>" +
" <config/>\n" +
" </edit-config>\n" +
"</rpc>");
}
private static final String MSG_ID_PLACEHOLDER_REGEX = "\\{MSG_ID\\}";
- private static final String PHYS_ADDR_PLACEHOLDER_REGEX = "\\{PHYS_ADDR\\}";
+ private static final String PHYS_ADDR_PLACEHOLDER = "{PHYS_ADDR}";
+
+ private static long macStart = 0xAABBCCDD0000L;
public static void main(final String[] args) {
final Parameters params = parseArgs(args, Parameters.getParser());
params.validate();
- // Wait 5 seconds to allow for debugging/profiling
- try {
- Thread.sleep(5000);
- } catch (final InterruptedException e) {
- throw new RuntimeException(e);
- }
-
final ch.qos.logback.classic.Logger root = (ch.qos.logback.classic.Logger) LoggerFactory.getLogger(Logger.ROOT_LOGGER_NAME);
root.setLevel(params.debug ? Level.DEBUG : Level.INFO);
+ final int threadAmount = params.threadAmount;
+ LOG.info("thread amount: " + threadAmount);
+ final int requestsPerThread = params.editCount / params.threadAmount;
+ LOG.info("requestsPerThread: " + requestsPerThread);
+ final int leftoverRequests = params.editCount % params.threadAmount;
+ LOG.info("leftoverRequests: " + leftoverRequests);
+
+
LOG.info("Preparing messages");
// Prepare all msgs up front
- final List<NetconfMessage> preparedMessages = Lists.newArrayListWithCapacity(params.editCount);
+ final List<List<NetconfMessage>> allPreparedMessages = new ArrayList<>(threadAmount);
+ for (int i = 0; i < threadAmount; i++) {
+ if (i != threadAmount - 1) {
+ allPreparedMessages.add(new ArrayList<NetconfMessage>(requestsPerThread));
+ } else {
+ allPreparedMessages.add(new ArrayList<NetconfMessage>(requestsPerThread + leftoverRequests));
+ }
+ }
+
final String editContentString;
try {
editContentString = Files.toString(params.editContent, Charsets.UTF_8);
- } catch (IOException e) {
+ } catch (final IOException e) {
throw new IllegalArgumentException("Cannot read content of " + params.editContent);
}
- for (int i = 0; i < params.editCount; i++) {
- final Document msg = XmlUtil.createDocumentCopy(editBlueprint);
- msg.getDocumentElement().setAttribute("message-id", Integer.toString(i));
- final NetconfMessage netconfMessage = new NetconfMessage(msg);
-
- final Element editContentElement;
- try {
- // Insert message id where needed
- String specificEditContent =
- editContentString.replaceAll(MSG_ID_PLACEHOLDER_REGEX, Integer.toString(i));
-
- // Insert physical address where needed
- specificEditContent =
- specificEditContent.replaceAll(PHYS_ADDR_PLACEHOLDER_REGEX, getMac(i));
-
- editContentElement = XmlUtil.readXmlToElement(specificEditContent);
- final Node config = ((Element) msg.getDocumentElement().getElementsByTagName("edit-config").item(0)).
- getElementsByTagName("config").item(0);
- config.appendChild(msg.importNode(editContentElement, true));
- } catch (final IOException | SAXException e) {
- throw new IllegalArgumentException("Edit content file is unreadable", e);
+ for (int i = 0; i < threadAmount; i++) {
+ final List<NetconfMessage> preparedMessages = allPreparedMessages.get(i);
+ int padding = 0;
+ if (i == threadAmount - 1) {
+ padding = leftoverRequests;
+ }
+ for (int j = 0; j < requestsPerThread + padding; j++) {
+ LOG.debug("id: " + (i * requestsPerThread + j));
+ preparedMessages.add(prepareMessage(i * requestsPerThread + j, editContentString));
}
-
- preparedMessages.add(netconfMessage);
-
}
-
final NioEventLoopGroup nioGroup = new NioEventLoopGroup();
final Timer timer = new HashedWheelTimer();
final NetconfClientDispatcherImpl netconfClientDispatcher = configureClientDispatcher(params, nioGroup, timer);
- final NetconfDeviceCommunicator sessionListener = getSessionListener(params.getInetAddress());
+ final List<StressClientCallable> callables = new ArrayList<>(threadAmount);
+ for (final List<NetconfMessage> messages : allPreparedMessages) {
+ callables.add(new StressClientCallable(params, netconfClientDispatcher, messages));
+ }
- final NetconfClientConfiguration cfg = getNetconfClientConfiguration(params, sessionListener);
+ final ExecutorService executorService = Executors.newFixedThreadPool(threadAmount);
- LOG.info("Connecting to netconf server {}:{}", params.ip, params.port);
- final NetconfClientSession netconfClientSession;
+ LOG.info("Starting stress test");
+ final Stopwatch started = Stopwatch.createStarted();
try {
- netconfClientSession = netconfClientDispatcher.createClient(cfg).get();
+ final List<Future<Boolean>> futures = executorService.invokeAll(callables);
+ for (final Future<Boolean> future : futures) {
+ try {
+ future.get(4L, TimeUnit.MINUTES);
+ } catch (ExecutionException | TimeoutException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ executorService.shutdownNow();
} catch (final InterruptedException e) {
- throw new RuntimeException(e);
- } catch (final ExecutionException e) {
- throw new RuntimeException("Unable to connect", e);
+ throw new RuntimeException("Unable to execute requests", e);
}
-
- LOG.info("Starting stress test");
- final Stopwatch started = Stopwatch.createStarted();
- getExecutionStrategy(params, preparedMessages, sessionListener).invoke();
started.stop();
LOG.info("FINISHED. Execution time: {}", started);
LOG.info("Requests per second: {}", (params.editCount * 1000.0 / started.elapsed(TimeUnit.MILLISECONDS)));
// Cleanup
- netconfClientSession.close();
timer.stop();
try {
nioGroup.shutdownGracefully().get(20L, TimeUnit.SECONDS);
}
}
- private static String getMac(final int i) {
- final String hex = Integer.toHexString(i);
- final Iterable<String> macGroups = Splitter.fixedLength(2).split(hex);
+ static NetconfMessage prepareMessage(final int id, final String editContentString) {
+ final Document msg = XmlUtil.createDocumentCopy(editBlueprint);
+ msg.getDocumentElement().setAttribute("message-id", Integer.toString(id));
+ final NetconfMessage netconfMessage = new NetconfMessage(msg);
- final int additional = 6 - Iterables.size(macGroups);
- final ArrayList<String> additionalGroups = Lists.newArrayListWithCapacity(additional);
- for (int j = 0; j < additional; j++) {
- additionalGroups.add("00");
- }
- return Joiner.on(':').join(Iterables.concat(Iterables.transform(macGroups, new Function<String, String>() {
- @Override
- public String apply(final String input) {
- return input.length() == 1 ? input + "0" : input;
+ final Element editContentElement;
+ try {
+ // Insert message id where needed
+ String specificEditContent = editContentString.replaceAll(MSG_ID_PLACEHOLDER_REGEX, Integer.toString(id));
+
+ final StringBuilder stringBuilder = new StringBuilder(specificEditContent);
+ int idx = stringBuilder.indexOf(PHYS_ADDR_PLACEHOLDER);
+ while (idx!= -1) {
+ stringBuilder.replace(idx, idx + PHYS_ADDR_PLACEHOLDER.length(), getMac(macStart++));
+ idx = stringBuilder.indexOf(PHYS_ADDR_PLACEHOLDER);
}
- }), additionalGroups));
- }
-
- private static ExecutionStrategy getExecutionStrategy(final Parameters params, final List<NetconfMessage> preparedMessages, final NetconfDeviceCommunicator sessionListener) {
- if(params.async) {
- return new AsyncExecutionStrategy(params, preparedMessages, sessionListener);
- } else {
- return new SyncExecutionStrategy(params, preparedMessages, sessionListener);
+ specificEditContent = stringBuilder.toString();
+
+ editContentElement = XmlUtil.readXmlToElement(specificEditContent);
+ final Node config = ((Element) msg.getDocumentElement().getElementsByTagName("edit-config").item(0)).
+ getElementsByTagName("config").item(0);
+ config.appendChild(msg.importNode(editContentElement, true));
+ } catch (final IOException | SAXException e) {
+ throw new IllegalArgumentException("Edit content file is unreadable", e);
}
+
+ return netconfMessage;
}
private static NetconfClientDispatcherImpl configureClientDispatcher(final Parameters params, final NioEventLoopGroup nioGroup, final Timer timer) {
return netconfClientDispatcher;
}
- private static NetconfClientConfiguration getNetconfClientConfiguration(final Parameters params, final NetconfDeviceCommunicator sessionListener) {
- final NetconfClientConfigurationBuilder netconfClientConfigurationBuilder = NetconfClientConfigurationBuilder.create();
- netconfClientConfigurationBuilder.withSessionListener(sessionListener);
- netconfClientConfigurationBuilder.withAddress(params.getInetAddress());
- if(params.tcpHeader != null) {
- final String header = params.tcpHeader.replaceAll("\"", "").trim() + "\n";
- netconfClientConfigurationBuilder.withAdditionalHeader(new NetconfHelloMessageAdditionalHeader(null, null, null, null, null) {
- @Override
- public String toFormattedString() {
- LOG.debug("Sending TCP header {}", header);
- return header;
- }
- });
+ public static String getMac(long mac) {
+ StringBuilder m = new StringBuilder(Long.toString(mac, 16));
+
+ for (int i = m.length(); i < 12; i++) {
+ m.insert(0, "0");
+ }
+
+ for (int j = m.length() - 2; j >= 2; j-=2) {
+ m.insert(j, ":");
}
- netconfClientConfigurationBuilder.withProtocol(params.ssh ? NetconfClientConfiguration.NetconfClientProtocol.SSH : NetconfClientConfiguration.NetconfClientProtocol.TCP);
- netconfClientConfigurationBuilder.withConnectionTimeoutMillis(20000L);
- netconfClientConfigurationBuilder.withReconnectStrategy(new NeverReconnectStrategy(GlobalEventExecutor.INSTANCE, 5000));
- return netconfClientConfigurationBuilder.build();
- }
- static NetconfDeviceCommunicator getSessionListener(final InetSocketAddress inetAddress) {
- final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> loggingRemoteDevice = new LoggingRemoteDevice();
- return new NetconfDeviceCommunicator(new RemoteDeviceId("secure-test", inetAddress), loggingRemoteDevice);
+ return m.toString();
}
private static Parameters parseArgs(final String[] args, final ArgumentParser parser) {
}
- private static class LoggingRemoteDevice implements RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> {
+ static class LoggingRemoteDevice implements RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> {
@Override
public void onRemoteSessionUp(final NetconfSessionPreferences remoteSessionCapabilities, final NetconfDeviceCommunicator netconfDeviceCommunicator) {
LOG.info("Session established");
--- /dev/null
+/*
+ * Copyright (c) 2015 Cisco Systems, Inc. and others. All rights reserved.
+ *
+ * This program and the accompanying materials are made available under the
+ * terms of the Eclipse Public License v1.0 which accompanies this distribution,
+ * and is available at http://www.eclipse.org/legal/epl-v10.html
+ */
+
+package org.opendaylight.controller.netconf.test.tool.client.stress;
+
+import io.netty.util.concurrent.GlobalEventExecutor;
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import org.opendaylight.controller.netconf.api.NetconfMessage;
+import org.opendaylight.controller.netconf.client.NetconfClientDispatcherImpl;
+import org.opendaylight.controller.netconf.client.NetconfClientSession;
+import org.opendaylight.controller.netconf.client.conf.NetconfClientConfiguration;
+import org.opendaylight.controller.netconf.client.conf.NetconfClientConfigurationBuilder;
+import org.opendaylight.controller.netconf.util.messages.NetconfHelloMessageAdditionalHeader;
+import org.opendaylight.controller.sal.connect.api.RemoteDevice;
+import org.opendaylight.controller.sal.connect.netconf.listener.NetconfDeviceCommunicator;
+import org.opendaylight.controller.sal.connect.netconf.listener.NetconfSessionPreferences;
+import org.opendaylight.controller.sal.connect.util.RemoteDeviceId;
+import org.opendaylight.protocol.framework.NeverReconnectStrategy;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class StressClientCallable implements Callable<Boolean>{
+
+ private static final Logger LOG = LoggerFactory.getLogger(StressClientCallable.class);
+
+ private Parameters params;
+ private final NetconfDeviceCommunicator sessionListener;
+ private final NetconfClientDispatcherImpl netconfClientDispatcher;
+ private final NetconfClientConfiguration cfg;
+ private final NetconfClientSession netconfClientSession;
+ private final ExecutionStrategy executionStrategy;
+
+ public StressClientCallable(final Parameters params,
+ final NetconfClientDispatcherImpl netconfClientDispatcher,
+ final List<NetconfMessage> preparedMessages) {
+ this.params = params;
+ this.sessionListener = getSessionListener(params.getInetAddress());
+ this.netconfClientDispatcher = netconfClientDispatcher;
+ cfg = getNetconfClientConfiguration(this.params, this.sessionListener);
+
+ LOG.info("Connecting to netconf server {}:{}", params.ip, params.port);
+ try {
+ netconfClientSession = netconfClientDispatcher.createClient(cfg).get();
+ } catch (final InterruptedException e) {
+ throw new RuntimeException(e);
+ } catch (final ExecutionException e) {
+ throw new RuntimeException("Unable to connect", e);
+ }
+ executionStrategy = getExecutionStrategy(params, preparedMessages, sessionListener);
+ }
+
+ @Override
+ public Boolean call() throws Exception {
+ executionStrategy.invoke();
+ netconfClientSession.close();
+ return true;
+ }
+
+ private static ExecutionStrategy getExecutionStrategy(final Parameters params, final List<NetconfMessage> preparedMessages, final NetconfDeviceCommunicator sessionListener) {
+ if(params.async) {
+ return new AsyncExecutionStrategy(params, preparedMessages, sessionListener);
+ } else {
+ return new SyncExecutionStrategy(params, preparedMessages, sessionListener);
+ }
+ }
+
+ private static NetconfDeviceCommunicator getSessionListener(final InetSocketAddress inetAddress) {
+ final RemoteDevice<NetconfSessionPreferences, NetconfMessage, NetconfDeviceCommunicator> loggingRemoteDevice = new StressClient.LoggingRemoteDevice();
+ return new NetconfDeviceCommunicator(new RemoteDeviceId("secure-test", inetAddress), loggingRemoteDevice);
+ }
+
+ private static NetconfClientConfiguration getNetconfClientConfiguration(final Parameters params, final NetconfDeviceCommunicator sessionListener) {
+ final NetconfClientConfigurationBuilder netconfClientConfigurationBuilder = NetconfClientConfigurationBuilder.create();
+ netconfClientConfigurationBuilder.withSessionListener(sessionListener);
+ netconfClientConfigurationBuilder.withAddress(params.getInetAddress());
+ if(params.tcpHeader != null) {
+ final String header = params.tcpHeader.replaceAll("\"", "").trim() + "\n";
+ netconfClientConfigurationBuilder.withAdditionalHeader(new NetconfHelloMessageAdditionalHeader(null, null, null, null, null) {
+ @Override
+ public String toFormattedString() {
+ LOG.debug("Sending TCP header {}", header);
+ return header;
+ }
+ });
+ }
+ netconfClientConfigurationBuilder.withProtocol(params.ssh ? NetconfClientConfiguration.NetconfClientProtocol.SSH : NetconfClientConfiguration.NetconfClientProtocol.TCP);
+ netconfClientConfigurationBuilder.withConnectionTimeoutMillis(20000L);
+ netconfClientConfigurationBuilder.withReconnectStrategy(new NeverReconnectStrategy(GlobalEventExecutor.INSTANCE, 5000));
+ return netconfClientConfigurationBuilder.build();
+ }
+}
private final List<NetconfMessage> preparedMessages;
private final NetconfDeviceCommunicator sessionListener;
private final List<Integer> editBatches;
+ private final int editAmount;
public SyncExecutionStrategy(final Parameters params, final List<NetconfMessage> preparedMessages, final NetconfDeviceCommunicator sessionListener) {
this.params = params;
this.preparedMessages = preparedMessages;
this.sessionListener = sessionListener;
- editBatches = countEditBatchSizes(params);
+ this.editBatches = countEditBatchSizes(params, preparedMessages.size());
+ editAmount = preparedMessages.size();
}
- private static List<Integer> countEditBatchSizes(final Parameters params) {
+ private static List<Integer> countEditBatchSizes(final Parameters params, final int amount) {
final List<Integer> editBatches = Lists.newArrayList();
- if (params.editBatchSize != params.editCount) {
- final int fullBatches = params.editCount / params.editBatchSize;
+ if (params.editBatchSize != amount) {
+ final int fullBatches = amount / params.editBatchSize;
for (int i = 0; i < fullBatches; i++) {
editBatches.add(params.editBatchSize);
}
- if (params.editCount % params.editBatchSize != 0) {
- editBatches.add(params.editCount % params.editBatchSize);
+ if (amount % params.editBatchSize != 0) {
+ editBatches.add(amount % params.editBatchSize);
}
} else {
editBatches.add(params.editBatchSize);
sessionListener.sendRequest(StressClient.COMMIT_MSG, StressClient.COMMIT_QNAME));
}
- Preconditions.checkState(responseCounter.get() == params.editCount + editBatches.size(), "Not all responses were received, only %s from %s", responseCounter.get(), params.editCount + editBatches.size());
+ Preconditions.checkState(responseCounter.get() == editAmount + editBatches.size(), "Not all responses were received, only %s from %s", responseCounter.get(), params.editCount + editBatches.size());
}
private void waitForResponse(AtomicInteger responseCounter, final ListenableFuture<RpcResult<NetconfMessage>> netconfMessageFuture) {
Code Review / controller.git / commitdiff