<properties>
<features.file>features.xml</features.file>
+ <org.json.version>20131018</org.json.version>
</properties>
<dependencies>
<type>xml</type>
<classifier>config</classifier>
</dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller.samples</groupId>
+ <artifactId>clustering-it-config</artifactId>
+ <version>${mdsal.version}</version>
+ <type>xml</type>
+ <classifier>testmoduleshardconf</classifier>
+ </dependency>
+ <dependency>
+ <groupId>org.opendaylight.controller.samples</groupId>
+ <artifactId>clustering-it-config</artifactId>
+ <version>${mdsal.version}</version>
+ <type>xml</type>
+ <classifier>testmoduleconf</classifier>
+ </dependency>
<dependency>
<groupId>org.opendaylight.controller</groupId>
<artifactId>sal-rest-docgen</artifactId>
*/
package org.opendaylight.controller.md.sal.dom.store.benchmark;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
import org.opendaylight.controller.md.sal.dom.store.impl.InMemoryDOMDataStore;
import org.opendaylight.yangtools.util.concurrent.SpecialExecutors;
-import org.openjdk.jmh.annotations.Level;
-import org.openjdk.jmh.annotations.Setup;
-import org.openjdk.jmh.annotations.TearDown;
-import org.openjdk.jmh.annotations.Fork;
-import org.openjdk.jmh.annotations.Scope;
-import org.openjdk.jmh.annotations.State;
import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
import org.openjdk.jmh.annotations.Mode;
import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.TearDown;
/**
* Benchmark for testing of performance of write operations for InMemoryDataStore. The instance
private static final int MAX_DATA_CHANGE_EXECUTOR_QUEUE_SIZE = 1000;
private static final int MAX_DATA_STORE_EXECUTOR_QUEUE_SIZE = 5000;
+ @Override
@Setup(Level.Trial)
public void setUp() throws Exception {
final String name = "DS_BENCHMARK";
final ExecutorService dataChangeListenerExecutor = SpecialExecutors.newBlockingBoundedFastThreadPool(
MAX_DATA_CHANGE_EXECUTOR_POOL_SIZE, MAX_DATA_CHANGE_EXECUTOR_QUEUE_SIZE, name + "-DCL");
- final ExecutorService domStoreExecutor = SpecialExecutors.newBoundedSingleThreadExecutor(
- MAX_DATA_STORE_EXECUTOR_QUEUE_SIZE, "DOMStore-" + name );
+ final ListeningExecutorService domStoreExecutor = MoreExecutors.listeningDecorator(SpecialExecutors.newBoundedSingleThreadExecutor(
+ MAX_DATA_STORE_EXECUTOR_QUEUE_SIZE, "DOMStore-" + name ));
domStore = new InMemoryDOMDataStore(name, domStoreExecutor,
dataChangeListenerExecutor);
initTestNode();
}
+ @Override
@TearDown
public void tearDown() {
schemaContext = null;
import akka.actor.ActorRef;
import akka.actor.Props;
import akka.testkit.JavaTestKit;
+import akka.util.Timeout;
+import com.google.protobuf.ByteString;
import junit.framework.Assert;
import org.junit.Test;
import org.opendaylight.controller.cluster.raft.DefaultConfigParamsImpl;
import org.opendaylight.controller.cluster.raft.RaftActorContext;
import org.opendaylight.controller.cluster.raft.RaftState;
import org.opendaylight.controller.cluster.raft.ReplicatedLogEntry;
+import org.opendaylight.controller.cluster.raft.base.messages.ApplySnapshot;
import org.opendaylight.controller.cluster.raft.base.messages.ElectionTimeout;
import org.opendaylight.controller.cluster.raft.messages.AppendEntries;
import org.opendaylight.controller.cluster.raft.messages.AppendEntriesReply;
+import org.opendaylight.controller.cluster.raft.messages.InstallSnapshot;
+import org.opendaylight.controller.cluster.raft.messages.InstallSnapshotReply;
import org.opendaylight.controller.cluster.raft.messages.RequestVote;
import org.opendaylight.controller.cluster.raft.messages.RequestVoteReply;
import org.opendaylight.controller.cluster.raft.utils.DoNothingActor;
-
+import org.opendaylight.controller.cluster.raft.utils.MessageCollectorActor;
+import scala.concurrent.Await;
+import scala.concurrent.Future;
+import scala.concurrent.duration.Duration;
+import scala.concurrent.duration.FiniteDuration;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import static akka.pattern.Patterns.ask;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
public class FollowerTest extends AbstractRaftActorBehaviorTest {
return new Follower(actorContext);
}
- @Override protected RaftActorContext createActorContext() {
- return new MockRaftActorContext("test", getSystem(), followerActor);
+ @Override protected RaftActorContext createActorContext() {
+ return createActorContext(followerActor);
+ }
+
+ protected RaftActorContext createActorContext(ActorRef actorRef){
+ return new MockRaftActorContext("test", getSystem(), actorRef);
}
@Test
createActorContext();
context.setLastApplied(100);
- setLastLogEntry((MockRaftActorContext) context, 1, 100, new MockRaftActorContext.MockPayload(""));
+ setLastLogEntry((MockRaftActorContext) context, 1, 100,
+ new MockRaftActorContext.MockPayload(""));
((MockRaftActorContext) context).getReplicatedLog().setSnapshotIndex(99);
List<ReplicatedLogEntry> entries =
Arrays.asList(
- (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(2, 101,
- new MockRaftActorContext.MockPayload("foo"))
+ (ReplicatedLogEntry) new MockRaftActorContext.MockReplicatedLogEntry(2, 101,
+ new MockRaftActorContext.MockPayload("foo"))
);
// The new commitIndex is 101
}};
}
+
+ /**
+ * This test verifies that when InstallSnapshot is received by
+ * the follower its applied correctly.
+ *
+ * @throws Exception
+ */
+ @Test
+ public void testHandleInstallSnapshot() throws Exception {
+ JavaTestKit javaTestKit = new JavaTestKit(getSystem()) {{
+
+ ActorRef leaderActor = getSystem().actorOf(Props.create(
+ MessageCollectorActor.class));
+
+ MockRaftActorContext context = (MockRaftActorContext)
+ createActorContext(getRef());
+
+ Follower follower = (Follower)createBehavior(context);
+
+ HashMap<String, String> followerSnapshot = new HashMap<>();
+ followerSnapshot.put("1", "A");
+ followerSnapshot.put("2", "B");
+ followerSnapshot.put("3", "C");
+
+ ByteString bsSnapshot = toByteString(followerSnapshot);
+ ByteString chunkData = ByteString.EMPTY;
+ int offset = 0;
+ int snapshotLength = bsSnapshot.size();
+ int i = 1;
+
+ do {
+ chunkData = getNextChunk(bsSnapshot, offset);
+ final InstallSnapshot installSnapshot =
+ new InstallSnapshot(1, "leader-1", i, 1,
+ chunkData, i, 3);
+ follower.handleMessage(leaderActor, installSnapshot);
+ offset = offset + 50;
+ i++;
+ } while ((offset+50) < snapshotLength);
+
+ final InstallSnapshot installSnapshot3 = new InstallSnapshot(1, "leader-1", 3, 1, chunkData, 3, 3);
+ follower.handleMessage(leaderActor, installSnapshot3);
+
+ String[] matches = new ReceiveWhile<String>(String.class, duration("2 seconds")) {
+ @Override
+ protected String match(Object o) throws Exception {
+ if (o instanceof ApplySnapshot) {
+ ApplySnapshot as = (ApplySnapshot)o;
+ if (as.getSnapshot().getLastIndex() != installSnapshot3.getLastIncludedIndex()) {
+ return "applySnapshot-lastIndex-mismatch";
+ }
+ if (as.getSnapshot().getLastAppliedTerm() != installSnapshot3.getLastIncludedTerm()) {
+ return "applySnapshot-lastAppliedTerm-mismatch";
+ }
+ if (as.getSnapshot().getLastAppliedIndex() != installSnapshot3.getLastIncludedIndex()) {
+ return "applySnapshot-lastAppliedIndex-mismatch";
+ }
+ if (as.getSnapshot().getLastTerm() != installSnapshot3.getLastIncludedTerm()) {
+ return "applySnapshot-lastTerm-mismatch";
+ }
+ return "applySnapshot";
+ }
+
+ return "ignoreCase";
+ }
+ }.get();
+
+ String applySnapshotMatch = "";
+ for (String reply: matches) {
+ if (reply.startsWith("applySnapshot")) {
+ applySnapshotMatch = reply;
+ }
+ }
+
+ assertEquals("applySnapshot", applySnapshotMatch);
+
+ Object messages = executeLocalOperation(leaderActor, "get-all-messages");
+
+ assertNotNull(messages);
+ assertTrue(messages instanceof List);
+ List<Object> listMessages = (List<Object>) messages;
+
+ int installSnapshotReplyReceivedCount = 0;
+ for (Object message: listMessages) {
+ if (message instanceof InstallSnapshotReply) {
+ ++installSnapshotReplyReceivedCount;
+ }
+ }
+
+ assertEquals(3, installSnapshotReplyReceivedCount);
+
+ }};
+ }
+
+ public Object executeLocalOperation(ActorRef actor, Object message) throws Exception {
+ FiniteDuration operationDuration = Duration.create(5, TimeUnit.SECONDS);
+ Timeout operationTimeout = new Timeout(operationDuration);
+ Future<Object> future = ask(actor, message, operationTimeout);
+
+ try {
+ return Await.result(future, operationDuration);
+ } catch (Exception e) {
+ throw e;
+ }
+ }
+
+ public ByteString getNextChunk (ByteString bs, int offset){
+ int snapshotLength = bs.size();
+ int start = offset;
+ int size = 50;
+ if (50 > snapshotLength) {
+ size = snapshotLength;
+ } else {
+ if ((start + 50) > snapshotLength) {
+ size = snapshotLength - start;
+ }
+ }
+ return bs.substring(start, start + size);
+ }
+
+ private ByteString toByteString(Map<String, String> state) {
+ ByteArrayOutputStream b = null;
+ ObjectOutputStream o = null;
+ try {
+ try {
+ b = new ByteArrayOutputStream();
+ o = new ObjectOutputStream(b);
+ o.writeObject(state);
+ byte[] snapshotBytes = b.toByteArray();
+ return ByteString.copyFrom(snapshotBytes);
+ } finally {
+ if (o != null) {
+ o.flush();
+ o.close();
+ }
+ if (b != null) {
+ b.close();
+ }
+ }
+ } catch (IOException e) {
+ org.junit.Assert.fail("IOException in converting Hashmap to Bytestring:" + e);
+ }
+ return null;
+ }
}
--- /dev/null
+/*
+ * Copyright (c) 2013 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.cluster.raft.utils;
+
+import akka.actor.UntypedActor;
+
+import java.util.ArrayList;
+import java.util.List;
+
+
+public class MessageCollectorActor extends UntypedActor {
+ private List<Object> messages = new ArrayList<>();
+
+ @Override public void onReceive(Object message) throws Exception {
+ if(message instanceof String){
+ if("get-all-messages".equals(message)){
+ getSender().tell(messages, getSelf());
+ }
+ } else {
+ messages.add(message);
+ }
+ }
+}
import com.google.common.base.Preconditions;
import com.google.common.collect.ArrayListMultimap;
import com.google.common.collect.Multimap;
-
import java.util.Collection;
import java.util.Map.Entry;
-
import org.opendaylight.controller.md.sal.common.api.data.AsyncDataBroker.DataChangeScope;
import org.opendaylight.controller.md.sal.dom.store.impl.DOMImmutableDataChangeEvent.Builder;
import org.opendaylight.controller.md.sal.dom.store.impl.DOMImmutableDataChangeEvent.SimpleEventFactory;
Preconditions.checkArgument(node.getDataAfter().isPresent(),
"Modification at {} has type {} but no after-data", state.getPath(), node.getModificationType());
if (!node.getDataBefore().isPresent()) {
- resolveCreateEvent(state, node.getDataAfter().get());
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ final NormalizedNode<PathArgument, ?> afterNode = (NormalizedNode)node.getDataAfter().get();
+ resolveSameEventRecursivelly(state, afterNode, DOMImmutableDataChangeEvent.getCreateEventFactory());
return true;
}
case DELETE:
Preconditions.checkArgument(node.getDataBefore().isPresent(),
"Modification at {} has type {} but no before-data", state.getPath(), node.getModificationType());
- resolveDeleteEvent(state, node.getDataBefore().get());
+
+ @SuppressWarnings({ "unchecked", "rawtypes" })
+ final NormalizedNode<PathArgument, ?> beforeNode = (NormalizedNode)node.getDataBefore().get();
+ resolveSameEventRecursivelly(state, beforeNode, DOMImmutableDataChangeEvent.getRemoveEventFactory());
return true;
case UNMODIFIED:
return false;
return true;
}
- /**
- * Resolves create events deep down the interest listener tree.
- *
- * @param path
- * @param listeners
- * @param afterState
- * @return
- */
- private void resolveCreateEvent(final ResolveDataChangeState state, final NormalizedNode<?, ?> afterState) {
- @SuppressWarnings({ "unchecked", "rawtypes" })
- final NormalizedNode<PathArgument, ?> node = (NormalizedNode) afterState;
- resolveSameEventRecursivelly(state, node, DOMImmutableDataChangeEvent.getCreateEventFactory());
- }
-
- private void resolveDeleteEvent(final ResolveDataChangeState state, final NormalizedNode<?, ?> beforeState) {
- @SuppressWarnings({ "unchecked", "rawtypes" })
- final NormalizedNode<PathArgument, ?> node = (NormalizedNode) beforeState;
- resolveSameEventRecursivelly(state, node, DOMImmutableDataChangeEvent.getRemoveEventFactory());
- }
-
private void resolveSameEventRecursivelly(final ResolveDataChangeState state,
final NormalizedNode<PathArgument, ?> node, final SimpleEventFactory eventFactory) {
if (!state.needsProcessing()) {
Preconditions.checkArgument(modification.getDataBefore().isPresent(), "Subtree change with before-data not present at path %s", state.getPath());
Preconditions.checkArgument(modification.getDataAfter().isPresent(), "Subtree change with after-data not present at path %s", state.getPath());
+ if (!state.needsProcessing()) {
+ LOG.trace("Not processing modified subtree {}", state.getPath());
+ return true;
+ }
+
DataChangeScope scope = null;
for (DataTreeCandidateNode childMod : modification.getChildNodes()) {
final ResolveDataChangeState childState = state.child(childMod.getIdentifier());
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Iterables;
import com.google.common.collect.Sets;
-
+import java.net.URI;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
-
import org.opendaylight.controller.netconf.client.NetconfClientSession;
import org.opendaylight.controller.sal.connect.netconf.util.NetconfMessageTransformUtil;
import org.opendaylight.yangtools.yang.common.QName;
return fromStrings(session.getServerCapabilities());
}
- private static final QName cachedQName(String namespace, String revision, String moduleName) {
+ private static QName cachedQName(final String namespace, final String revision, final String moduleName) {
return QName.cachedReference(QName.create(namespace, revision, moduleName));
}
+ private static QName cachedQName(final String namespace, final String moduleName) {
+ return QName.cachedReference(QName.create(URI.create(namespace), null, moduleName).withoutRevision());
+ }
+
public static NetconfSessionCapabilities fromStrings(final Collection<String> capabilities) {
final Set<QName> moduleBasedCaps = new HashSet<>();
final Set<String> nonModuleCaps = Sets.newHashSet(capabilities);
String revision = REVISION_PARAM.from(queryParams);
if (revision != null) {
- moduleBasedCaps.add(cachedQName(namespace, revision, moduleName));
- nonModuleCaps.remove(capability);
+ addModuleQName(moduleBasedCaps, nonModuleCaps, capability, cachedQName(namespace, revision, moduleName));
continue;
}
* We have seen devices which mis-escape revision, but the revision may not
* even be there. First check if there is a substring that matches revision.
*/
- if (!Iterables.any(queryParams, CONTAINS_REVISION)) {
+ if (Iterables.any(queryParams, CONTAINS_REVISION)) {
+
+ LOG.debug("Netconf device was not reporting revision correctly, trying to get amp;revision=");
+ revision = BROKEN_REVISON_PARAM.from(queryParams);
+ if (revision == null) {
+ LOG.warn("Netconf device returned revision incorrectly escaped for {}, ignoring it", capability);
+ addModuleQName(moduleBasedCaps, nonModuleCaps, capability, cachedQName(namespace, moduleName));
+ } else {
+ addModuleQName(moduleBasedCaps, nonModuleCaps, capability, cachedQName(namespace, revision, moduleName));
+ }
continue;
}
- LOG.debug("Netconf device was not reporting revision correctly, trying to get amp;revision=");
- revision = BROKEN_REVISON_PARAM.from(queryParams);
- if (revision == null) {
- LOG.warn("Netconf device returned revision incorrectly escaped for {}, ignoring it", capability);
- }
-
- // FIXME: do we really want to continue here?
- moduleBasedCaps.add(cachedQName(namespace, revision, moduleName));
- nonModuleCaps.remove(capability);
+ // Fallback, no revision provided for module
+ addModuleQName(moduleBasedCaps, nonModuleCaps, capability, cachedQName(namespace, moduleName));
}
return new NetconfSessionCapabilities(ImmutableSet.copyOf(nonModuleCaps), ImmutableSet.copyOf(moduleBasedCaps));
}
+
+
+ private static void addModuleQName(final Set<QName> moduleBasedCaps, final Set<String> nonModuleCaps, final String capability, final QName qName) {
+ moduleBasedCaps.add(qName);
+ nonModuleCaps.remove(capability);
+ }
}
assertThat(merged.getNonModuleCaps(), JUnitMatchers.hasItem("urn:ietf:params:netconf:capability:rollback-on-error:1.0"));
}
+ @Test
+ public void testCapabilityNoRevision() throws Exception {
+ final List<String> caps1 = Lists.newArrayList(
+ "namespace:2?module=module2",
+ "namespace:2?module=module2&revision=2012-12-12",
+ "namespace:2?module=module1&RANDOMSTRING;revision=2013-12-12",
+ "namespace:2?module=module2&RANDOMSTRING;revision=2013-12-12" // This one should be ignored(same as first), since revision is in wrong format
+ );
+
+ final NetconfSessionCapabilities sessionCaps1 = NetconfSessionCapabilities.fromStrings(caps1);
+ assertCaps(sessionCaps1, 0, 3);
+ }
+
private void assertCaps(final NetconfSessionCapabilities sessionCaps1, final int nonModuleCaps, final int moduleCaps) {
assertEquals(nonModuleCaps, sessionCaps1.getNonModuleCaps().size());
assertEquals(moduleCaps, sessionCaps1.getModuleBasedCaps().size());
@GET
@Path("/modules")
- @Produces({ Draft02.MediaTypes.API + XML, Draft02.MediaTypes.API + JSON, MediaType.APPLICATION_JSON,
+ @Produces({ Draft02.MediaTypes.API + JSON, Draft02.MediaTypes.API + XML, MediaType.APPLICATION_JSON,
MediaType.APPLICATION_XML, MediaType.TEXT_XML })
public StructuredData getModules(@Context UriInfo uriInfo);
@GET
@Path("/modules/{identifier:.+}")
- @Produces({ Draft02.MediaTypes.API + XML, Draft02.MediaTypes.API + JSON, MediaType.APPLICATION_JSON,
+ @Produces({ Draft02.MediaTypes.API + JSON, Draft02.MediaTypes.API + XML, MediaType.APPLICATION_JSON,
MediaType.APPLICATION_XML, MediaType.TEXT_XML })
public StructuredData getModules(@PathParam("identifier") String identifier, @Context UriInfo uriInfo);
@GET
@Path("/modules/module/{identifier:.+}")
- @Produces({ Draft02.MediaTypes.API + XML, Draft02.MediaTypes.API + JSON, MediaType.APPLICATION_JSON,
+ @Produces({ Draft02.MediaTypes.API + JSON, Draft02.MediaTypes.API + XML, MediaType.APPLICATION_JSON,
MediaType.APPLICATION_XML, MediaType.TEXT_XML })
public StructuredData getModule(@PathParam("identifier") String identifier, @Context UriInfo uriInfo);
@GET
@Path("/operations")
- @Produces({ Draft02.MediaTypes.API + XML, Draft02.MediaTypes.API + JSON, MediaType.APPLICATION_JSON,
+ @Produces({ Draft02.MediaTypes.API + JSON, Draft02.MediaTypes.API + XML, MediaType.APPLICATION_JSON,
MediaType.APPLICATION_XML, MediaType.TEXT_XML })
public StructuredData getOperations(@Context UriInfo uriInfo);
@GET
@Path("/operations/{identifier:.+}")
- @Produces({ Draft02.MediaTypes.API + XML, Draft02.MediaTypes.API + JSON, MediaType.APPLICATION_JSON,
+ @Produces({ Draft02.MediaTypes.API + JSON, Draft02.MediaTypes.API + XML, MediaType.APPLICATION_JSON,
MediaType.APPLICATION_XML, MediaType.TEXT_XML })
public StructuredData getOperations(@PathParam("identifier") String identifier, @Context UriInfo uriInfo);
@GET
@Path("/streams")
- @Produces({ Draft02.MediaTypes.API + XML, Draft02.MediaTypes.API + JSON, MediaType.APPLICATION_JSON,
+ @Produces({ Draft02.MediaTypes.API + JSON, Draft02.MediaTypes.API + XML, MediaType.APPLICATION_JSON,
MediaType.APPLICATION_XML, MediaType.TEXT_XML })
public StructuredData getAvailableStreams(@Context UriInfo uriInfo);
LOG.debug("In toResponse: {}", exception.getMessage());
- // Default to the content type if there's no Accept header
- MediaType mediaType = headers.getMediaType();
List<MediaType> accepts = headers.getAcceptableMediaTypes();
+ accepts.remove(MediaType.WILDCARD_TYPE);
LOG.debug("Accept headers: {}", accepts);
+ final MediaType mediaType;
if (accepts != null && accepts.size() > 0) {
mediaType = accepts.get(0); // just pick the first one
+ } else {
+ // Default to the content type if there's no Accept header
+ mediaType = MediaType.APPLICATION_JSON_TYPE;
}
LOG.debug("Using MediaType: {}", mediaType);
<artifactId>config-util</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>sal-netconf-connector</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>netconf-api</artifactId>
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
import io.netty.channel.local.LocalAddress;
-import io.netty.util.concurrent.Future;
-import io.netty.util.concurrent.GenericFutureListener;
import io.netty.util.concurrent.GlobalEventExecutor;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
import org.opendaylight.controller.netconf.api.NetconfMessage;
import org.opendaylight.controller.netconf.auth.AuthProvider;
import org.opendaylight.controller.netconf.client.NetconfClientDispatcher;
import org.opendaylight.controller.netconf.client.NetconfClientDispatcherImpl;
+import org.opendaylight.controller.netconf.client.NetconfClientSessionListener;
import org.opendaylight.controller.netconf.client.SimpleNetconfClientSessionListener;
import org.opendaylight.controller.netconf.client.conf.NetconfClientConfiguration;
import org.opendaylight.controller.netconf.client.conf.NetconfClientConfigurationBuilder;
import org.opendaylight.controller.netconf.util.messages.NetconfMessageUtil;
import org.opendaylight.controller.netconf.util.osgi.NetconfConfigUtil;
import org.opendaylight.controller.netconf.util.xml.XmlUtil;
+import org.opendaylight.controller.sal.connect.api.RemoteDevice;
+import org.opendaylight.controller.sal.connect.api.RemoteDeviceCommunicator;
+import org.opendaylight.controller.sal.connect.netconf.listener.NetconfDeviceCommunicator;
+import org.opendaylight.controller.sal.connect.netconf.listener.NetconfSessionCapabilities;
+import org.opendaylight.controller.sal.connect.util.RemoteDeviceId;
import org.opendaylight.protocol.framework.NeverReconnectStrategy;
+import org.opendaylight.yangtools.yang.common.QName;
+import org.opendaylight.yangtools.yang.common.RpcResult;
+import org.xml.sax.SAXException;
public class NetconfITSecureTest extends AbstractNetconfConfigTest {
@Test
public void testSecure() throws Exception {
final NetconfClientDispatcher dispatch = new NetconfClientDispatcherImpl(getNettyThreadgroup(), getNettyThreadgroup(), getHashedWheelTimer());
- try (TestingNetconfClient netconfClient = new TestingNetconfClient("testing-ssh-client", dispatch, getClientConfiguration())) {
+ try (TestingNetconfClient netconfClient = new TestingNetconfClient("testing-ssh-client", dispatch, getClientConfiguration(new SimpleNetconfClientSessionListener()))) {
NetconfMessage response = netconfClient.sendMessage(getGetConfig());
assertFalse("Unexpected error message " + XmlUtil.toString(response.getDocument()),
NetconfMessageUtil.isErrorMessage(response));
/**
* Test all requests are handled properly and no mismatch occurs in listener
*/
- @Test(timeout = 3*60*1000)
+ @Test(timeout = 5*60*1000)
public void testSecureStress() throws Exception {
+ final int requests = 10000;
+
final NetconfClientDispatcher dispatch = new NetconfClientDispatcherImpl(getNettyThreadgroup(), getNettyThreadgroup(), getHashedWheelTimer());
- try (TestingNetconfClient netconfClient = new TestingNetconfClient("testing-ssh-client", dispatch, getClientConfiguration())) {
+ final NetconfDeviceCommunicator sessionListener = getSessionListener();
+ try (TestingNetconfClient netconfClient = new TestingNetconfClient("testing-ssh-client", dispatch, getClientConfiguration(sessionListener))) {
final AtomicInteger responseCounter = new AtomicInteger(0);
- final List<Future<?>> futures = Lists.newArrayList();
+ final List<ListenableFuture<RpcResult<NetconfMessage>>> futures = Lists.newArrayList();
- final int requests = 1000;
for (int i = 0; i < requests; i++) {
- final Future<NetconfMessage> netconfMessageFuture = netconfClient.sendRequest(getGetConfig());
+ NetconfMessage getConfig = getGetConfig();
+ getConfig = changeMessageId(getConfig, i);
+ final ListenableFuture<RpcResult<NetconfMessage>> netconfMessageFuture = sessionListener.sendRequest(getConfig, QName.create("namespace", "2012-12-12", "get"));
futures.add(netconfMessageFuture);
- netconfMessageFuture.addListener(new GenericFutureListener<Future<? super NetconfMessage>>() {
+ Futures.addCallback(netconfMessageFuture, new FutureCallback<RpcResult<NetconfMessage>>() {
@Override
- public void operationComplete(final Future<? super NetconfMessage> future) throws Exception {
- assertTrue("Request unsuccessful " + future.cause(), future.isSuccess());
+ public void onSuccess(final RpcResult<NetconfMessage> result) {
responseCounter.incrementAndGet();
}
+
+ @Override
+ public void onFailure(final Throwable t) {
+ throw new RuntimeException(t);
+ }
});
}
- for (final Future<?> future : futures) {
- future.await();
+ // Wait for every future
+ for (final ListenableFuture<RpcResult<NetconfMessage>> future : futures) {
+ try {
+ future.get(3, TimeUnit.MINUTES);
+ } catch (final TimeoutException e) {
+ fail("Request " + futures.indexOf(future) + " is not responding");
+ }
}
// Give future listeners some time to finish counter incrementation
}
}
- public NetconfClientConfiguration getClientConfiguration() throws IOException {
+ private NetconfMessage changeMessageId(final NetconfMessage getConfig, final int i) throws IOException, SAXException {
+ String s = XmlUtil.toString(getConfig.getDocument(), false);
+ s = s.replace("101", Integer.toString(i));
+ return new NetconfMessage(XmlUtil.readXmlToDocument(s));
+ }
+
+ public NetconfClientConfiguration getClientConfiguration(final NetconfClientSessionListener sessionListener) throws IOException {
final NetconfClientConfigurationBuilder b = NetconfClientConfigurationBuilder.create();
b.withAddress(TLS_ADDRESS);
- b.withSessionListener(new SimpleNetconfClientSessionListener());
+ // Using session listener from sal-netconf-connector since stress test cannot be performed with simple listener
+ b.withSessionListener(sessionListener);
b.withReconnectStrategy(new NeverReconnectStrategy(GlobalEventExecutor.INSTANCE, 5000));
b.withProtocol(NetconfClientConfiguration.NetconfClientProtocol.SSH);
b.withConnectionTimeoutMillis(5000);
return b.build();
}
+ @Mock
+ private RemoteDevice<NetconfSessionCapabilities, NetconfMessage> mockedRemoteDevice;
+
+ private NetconfDeviceCommunicator getSessionListener() {
+ MockitoAnnotations.initMocks(this);
+ doNothing().when(mockedRemoteDevice).onRemoteSessionUp(any(NetconfSessionCapabilities.class), any(RemoteDeviceCommunicator.class));
+ doNothing().when(mockedRemoteDevice).onRemoteSessionDown();
+ return new NetconfDeviceCommunicator(new RemoteDeviceId("secure-test"), mockedRemoteDevice);
+ }
+
public AuthProvider getAuthProvider() throws Exception {
final AuthProvider mockAuth = mock(AuthProvider.class);
doReturn("mockedAuth").when(mockAuth).toString();
</appender>
<logger name="org.opendaylight.controller.netconf" level="TRACE"/>
+ <logger name="org.opendaylight.controller.sal.connect.netconf" level="TRACE"/>
<root level="error">
<appender-ref ref="STDOUT" />
", remote window is not getting read or is too small"));
}
+ // We need to reset buffer read index, since we've already read it when we tried to write it the first time
+ ((ByteBuf) msg).resetReaderIndex();
logger.debug("Write pending to SSH remote on channel: {}, current pending count: {}", ctx.channel(), pendingWriteCounter);
// In case of pending, re-invoke write after pending is finished
lastWriteFuture.addListener(new SshFutureListener<IoWriteFuture>() {
@Override
public void operationComplete(final IoWriteFuture future) {
+ // FIXME possible minor race condition, we cannot guarantee that this callback when pending is finished will be executed first
+ // External thread could trigger write on this instance while we are on this line
+ // Verify
if (future.isWritten()) {
synchronized (SshWriteAsyncHandler.this) {
// Pending done, decrease counter
pendingWriteCounter--;
+ write(ctx, msg, promise);
}
- write(ctx, msg, promise);
} else {
// Cannot reschedule pending, fail
handlePendingFailed(ctx, e);
Code Review / controller.git / commitdiff