<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>
if (LocalServerChannel.class.equals(channelClass) == false) {
// makes no sense for LocalServer and produces warning
b.childOption(ChannelOption.SO_KEEPALIVE, true);
+ b.childOption(ChannelOption.TCP_NODELAY , true);
}
b.childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
customizeBootstrap(b);
package org.opendaylight.controller.sal.binding.api;
import java.util.EventListener;
-
import org.opendaylight.yangtools.yang.binding.Notification;
/**
* capture of this interface.
*
* @param <T> the interested notification type
+ * @deprecated Deprecated unused API.
*/
+@Deprecated
public interface NotificationListener<T extends Notification> extends EventListener {
/**
* Invoked to deliver a notification.
import java.util.EventListener;
import java.util.concurrent.ExecutorService;
-
import org.opendaylight.controller.md.sal.common.api.notify.NotificationPublishService;
import org.opendaylight.yangtools.concepts.ListenerRegistration;
import org.opendaylight.yangtools.yang.binding.Notification;
* Interface for a notification service that provides publish/subscribe capabilities for YANG
* modeled notifications. This interface is a combination of the {@link NotificationService} and
* {@link NotificationPublishService} interfaces.
+ *
+ * @deprecated Please use {@link org.opendaylight.controller.md.sal.binding.api.NotificationPublishService}.
*/
+@Deprecated
public interface NotificationProviderService extends NotificationService, NotificationPublishService<Notification> {
/**
* </pre>
* The <code>onStart</code> method will be invoked when someone publishes a <code>Start</code> notification and
* the <code>onStop</code> method will be invoked when someone publishes a <code>Stop</code> notification.
+ *
+ * @deprecated Please use {@link org.opendaylight.controller.md.sal.binding.api.NotificationService} instead.
*/
+@Deprecated
public interface NotificationService extends BindingAwareService {
/**
* Registers a generic listener implementation for a specified notification type.
LOG.debug("Create rpc registry and broker actors");
rpcRegistry =
- getContext().actorOf(Props.create(RpcRegistry.class).
+ getContext().actorOf(RpcRegistry.props().
withMailbox(config.getMailBoxName()), config.getRpcRegistryName());
rpcBroker =
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
+import java.util.Set;
+
import org.opendaylight.controller.remote.rpc.registry.gossip.Copier;
import org.opendaylight.controller.sal.connector.api.RpcRouter;
}
}
+ public Set<RpcRouter.RouteIdentifier<?, ?, ?>> getRoutes() {
+ return table.keySet();
+ }
+
public void addRoute(RpcRouter.RouteIdentifier<?,?,?> routeId){
table.put(routeId, System.currentTimeMillis());
}
package org.opendaylight.controller.remote.rpc.registry;
import akka.actor.ActorRef;
+import akka.actor.Props;
+import akka.japi.Creator;
import akka.japi.Option;
import akka.japi.Pair;
import com.google.common.base.Preconditions;
import org.opendaylight.controller.remote.rpc.registry.RpcRegistry.Messages.SetLocalRouter;
import org.opendaylight.controller.remote.rpc.registry.gossip.Bucket;
import org.opendaylight.controller.remote.rpc.registry.gossip.BucketStore;
+import org.opendaylight.controller.remote.rpc.registry.mbeans.RemoteRpcRegistryMXBean;
+import org.opendaylight.controller.remote.rpc.registry.mbeans.RemoteRpcRegistryMXBeanImpl;
import org.opendaylight.controller.sal.connector.api.RpcRouter;
import org.opendaylight.controller.sal.connector.api.RpcRouter.RouteIdentifier;
getLocalBucket().setData(new RoutingTable());
}
+ public static Props props() {
+ return Props.create(new RpcRegistryCreator());
+ }
+
@Override
protected void handleReceive(Object message) throws Exception {
//TODO: if sender is remote, reject message
}
}
}
+
+ private static class RpcRegistryCreator implements Creator<RpcRegistry> {
+ private static final long serialVersionUID = 1L;
+
+ @Override
+ public RpcRegistry create() throws Exception {
+ RpcRegistry registry = new RpcRegistry();
+ RemoteRpcRegistryMXBean mxBean = new RemoteRpcRegistryMXBeanImpl(registry);
+ return registry;
+ }
+ }
}
import akka.actor.Address;
import akka.actor.Props;
import akka.cluster.ClusterActorRefProvider;
-import com.google.common.annotations.VisibleForTesting;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
}
}
- protected BucketImpl<T> getLocalBucket() {
+ public BucketImpl<T> getLocalBucket() {
return localBucket;
}
versions.put(selfAddress, localBucket.getVersion());
}
- p
rotected Map<Address, Bucket<T>> getRemoteBuckets() {
+ p
ublic Map<Address, Bucket<T>> getRemoteBuckets() {
return remoteBuckets;
}
- @VisibleForTesting
- Map<Address, Long> getVersions() {
+ public Map<Address, Long> getVersions() {
return versions;
}
}
--- /dev/null
+package org.opendaylight.controller.remote.rpc.registry.mbeans;
+
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * JMX bean to check remote rpc registry
+ */
+
+public interface RemoteRpcRegistryMXBean {
+
+ Set<String> getGlobalRpc();
+
+ String getBucketVersions();
+
+ Set<String> getLocalRegisteredRoutedRpc();
+
+ Map<String,String> findRpcByName(String name);
+
+ Map<String,String> findRpcByRoute(String route);
+}
--- /dev/null
+package org.opendaylight.controller.remote.rpc.registry.mbeans;
+
+import akka.actor.Address;
+import org.opendaylight.controller.md.sal.common.util.jmx.AbstractMXBean;
+import org.opendaylight.controller.remote.rpc.registry.RoutingTable;
+import org.opendaylight.controller.remote.rpc.registry.RpcRegistry;
+import org.opendaylight.controller.remote.rpc.registry.gossip.Bucket;
+import org.opendaylight.controller.sal.connector.api.RpcRouter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+
+public class RemoteRpcRegistryMXBeanImpl extends AbstractMXBean implements RemoteRpcRegistryMXBean {
+
+ protected final Logger log = LoggerFactory.getLogger(getClass());
+
+ private final String NULL_CONSTANT = "null";
+
+ private final String LOCAL_CONSTANT = "local";
+
+ private final String ROUTE_CONSTANT = "route:";
+
+ private final String NAME_CONSTANT = " | name:";
+
+ private final RpcRegistry rpcRegistry;
+
+ public RemoteRpcRegistryMXBeanImpl(final RpcRegistry rpcRegistry) {
+ super("RemoteRpcRegistry", "RemoteRpcBroker", null);
+ this.rpcRegistry = rpcRegistry;
+ registerMBean();
+ }
+
+ @Override
+ public Set<String> getGlobalRpc() {
+ RoutingTable table = rpcRegistry.getLocalBucket().getData();
+ Set<String> globalRpc = new HashSet<>(table.getRoutes().size());
+ for(RpcRouter.RouteIdentifier<?, ?, ?> route : table.getRoutes()){
+ if(route.getRoute() == null) {
+ globalRpc.add(route.getType() != null ? route.getType().toString() : NULL_CONSTANT);
+ }
+ }
+ if(log.isDebugEnabled()) {
+ log.debug("Locally registered global RPCs {}", globalRpc);
+ }
+ return globalRpc;
+ }
+
+ @Override
+ public Set<String> getLocalRegisteredRoutedRpc() {
+ RoutingTable table = rpcRegistry.getLocalBucket().getData();
+ Set<String> routedRpc = new HashSet<>(table.getRoutes().size());
+ for(RpcRouter.RouteIdentifier<?, ?, ?> route : table.getRoutes()){
+ if(route.getRoute() != null) {
+ StringBuilder builder = new StringBuilder(ROUTE_CONSTANT);
+ builder.append(route.getRoute().toString()).append(NAME_CONSTANT).append(route.getType() != null ?
+ route.getType().toString() : NULL_CONSTANT);
+ routedRpc.add(builder.toString());
+ }
+ }
+ if(log.isDebugEnabled()) {
+ log.debug("Locally registered routed RPCs {}", routedRpc);
+ }
+ return routedRpc;
+ }
+
+ @Override
+ public Map<String, String> findRpcByName(final String name) {
+ RoutingTable localTable = rpcRegistry.getLocalBucket().getData();
+ // Get all RPCs from local bucket
+ Map<String, String> rpcMap = new HashMap<>(getRpcMemberMapByName(localTable, name, LOCAL_CONSTANT));
+
+ // Get all RPCs from remote bucket
+ Map<Address, Bucket<RoutingTable>> buckets = rpcRegistry.getRemoteBuckets();
+ for(Address address : buckets.keySet()) {
+ RoutingTable table = buckets.get(address).getData();
+ rpcMap.putAll(getRpcMemberMapByName(table, name, address.toString()));
+ }
+ if(log.isDebugEnabled()) {
+ log.debug("list of RPCs {} searched by name {}", rpcMap, name);
+ }
+ return rpcMap;
+ }
+
+ @Override
+ public Map<String, String> findRpcByRoute(String routeId) {
+ RoutingTable localTable = rpcRegistry.getLocalBucket().getData();
+ Map<String, String> rpcMap = new HashMap<>(getRpcMemberMapByRoute(localTable, routeId, LOCAL_CONSTANT));
+
+ Map<Address, Bucket<RoutingTable>> buckets = rpcRegistry.getRemoteBuckets();
+ for(Address address : buckets.keySet()) {
+ RoutingTable table = buckets.get(address).getData();
+ rpcMap.putAll(getRpcMemberMapByRoute(table, routeId, address.toString()));
+
+ }
+ if(log.isDebugEnabled()) {
+ log.debug("list of RPCs {} searched by route {}", rpcMap, routeId);
+ }
+ return rpcMap;
+ }
+
+ /**
+ * Search if the routing table route String contains routeName
+ */
+
+ private Map<String,String> getRpcMemberMapByRoute(final RoutingTable table, final String routeName,
+ final String address) {
+ Set<RpcRouter.RouteIdentifier<?, ?, ?>> routes = table.getRoutes();
+ Map<String, String> rpcMap = new HashMap<>(routes.size());
+ for(RpcRouter.RouteIdentifier<?, ?, ?> route : table.getRoutes()){
+ if(route.getRoute() != null) {
+ String routeString = route.getRoute().toString();
+ if(routeString.contains(routeName)) {
+ StringBuilder builder = new StringBuilder(ROUTE_CONSTANT);
+ builder.append(routeString).append(NAME_CONSTANT).append(route.getType() != null ?
+ route.getType().toString() : NULL_CONSTANT);
+ rpcMap.put(builder.toString(), address);
+ }
+ }
+ }
+ return rpcMap;
+ }
+
+ /**
+ * Search if the routing table route type contains name
+ */
+ private Map<String, String> getRpcMemberMapByName(final RoutingTable table, final String name,
+ final String address) {
+ Set<RpcRouter.RouteIdentifier<?, ?, ?>> routes = table.getRoutes();
+ Map<String, String> rpcMap = new HashMap<>(routes.size());
+ for(RpcRouter.RouteIdentifier<?, ?, ?> route : routes){
+ if(route.getType() != null) {
+ String type = route.getType().toString();
+ if(type.contains(name)) {
+ StringBuilder builder = new StringBuilder(ROUTE_CONSTANT);
+ builder.append(route.getRoute() != null ? route.getRoute().toString(): NULL_CONSTANT)
+ .append(NAME_CONSTANT).append(type);
+ rpcMap.put(builder.toString(), address);
+ }
+ }
+ }
+ return rpcMap;
+ }
+
+
+
+ @Override
+ public String getBucketVersions() {
+ return rpcRegistry.getVersions().toString();
+ }
+
+}
\ No newline at end of file
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);
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
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