Merge "Bug 1926: fixed features/mdsal/pom.xml dependencies"
authorTony Tkacik <ttkacik@cisco.com>
Wed, 17 Sep 2014 11:21:14 +0000 (11:21 +0000)
committerGerrit Code Review <gerrit@opendaylight.org>
Wed, 17 Sep 2014 11:21:14 +0000 (11:21 +0000)
12 files changed:
opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/behaviors/FollowerTest.java
opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/utils/MessageCollectorActor.java [new file with mode: 0644]
opendaylight/md-sal/sal-netconf-connector/src/main/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfSessionCapabilities.java
opendaylight/md-sal/sal-netconf-connector/src/test/java/org/opendaylight/controller/sal/connect/netconf/listener/NetconfSessionCapabilitiesTest.java
opendaylight/netconf/netconf-it/pom.xml
opendaylight/netconf/netconf-it/src/test/java/org/opendaylight/controller/netconf/it/NetconfITSecureTest.java
opendaylight/netconf/netconf-it/src/test/resources/logback-test.xml
opendaylight/netconf/netconf-netty-util/src/main/java/org/opendaylight/controller/netconf/nettyutil/handler/ssh/client/AsyncSshHandler.java
opendaylight/sal/api/src/main/java/org/opendaylight/controller/sal/packet/ICMP.java
opendaylight/sal/api/src/main/java/org/opendaylight/controller/sal/packet/IPv4.java
opendaylight/sal/api/src/test/java/org/opendaylight/controller/sal/packet/ICMPTest.java
opendaylight/sal/api/src/test/java/org/opendaylight/controller/sal/packet/IPv4Test.java

index 227d1ef..fd4a75a 100644 (file)
@@ -3,6 +3,8 @@ package org.opendaylight.controller.cluster.raft.behaviors;
 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;
@@ -10,19 +12,35 @@ import org.opendaylight.controller.cluster.raft.MockRaftActorContext;
 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 {
 
@@ -34,8 +52,12 @@ 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
@@ -158,13 +180,14 @@ public class FollowerTest extends AbstractRaftActorBehaviorTest {
                 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
@@ -409,4 +432,148 @@ public class FollowerTest extends AbstractRaftActorBehaviorTest {
         }};
     }
 
+
+    /**
+     * 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;
+    }
 }
diff --git a/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/utils/MessageCollectorActor.java b/opendaylight/md-sal/sal-akka-raft/src/test/java/org/opendaylight/controller/cluster/raft/utils/MessageCollectorActor.java
new file mode 100644 (file)
index 0000000..88eecfe
--- /dev/null
@@ -0,0 +1,29 @@
+/*
+ * 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);
+        }
+    }
+}
index 2642116..09e178f 100644 (file)
@@ -8,11 +8,10 @@ import com.google.common.base.Splitter;
 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;
@@ -119,10 +118,14 @@ public final class NetconfSessionCapabilities {
         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);
@@ -142,8 +145,7 @@ public final class NetconfSessionCapabilities {
 
             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;
             }
 
@@ -151,21 +153,29 @@ public final class NetconfSessionCapabilities {
              * 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);
+    }
 }
index 87947b5..80bb08f 100644 (file)
@@ -43,6 +43,19 @@ public class NetconfSessionCapabilitiesTest {
         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&amp;revision=2012-12-12",
+                "namespace:2?module=module1&amp;RANDOMSTRING;revision=2013-12-12",
+                "namespace:2?module=module2&amp;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());
index 272b686..3a70a39 100644 (file)
       <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>
index 4fe5f2a..bc8efbe 100644 (file)
@@ -10,27 +10,35 @@ package org.opendaylight.controller.netconf.it;
 
 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;
@@ -42,7 +50,15 @@ import org.opendaylight.controller.netconf.ssh.authentication.PEMGenerator;
 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 {
 
@@ -70,7 +86,7 @@ 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));
@@ -91,29 +107,42 @@ public class NetconfITSecureTest extends AbstractNetconfConfigTest {
     /**
      * 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
@@ -123,10 +152,17 @@ public class NetconfITSecureTest extends AbstractNetconfConfigTest {
         }
     }
 
-    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);
@@ -134,6 +170,16 @@ public class NetconfITSecureTest extends AbstractNetconfConfigTest {
         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();
index c5037d3..91fb805 100644 (file)
@@ -7,6 +7,7 @@
     </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" />
index 369c013..3d1e478 100644 (file)
@@ -328,6 +328,8 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
                             ", 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
@@ -335,12 +337,15 @@ public class AsyncSshHandler extends ChannelOutboundHandlerAdapter {
                 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);
index 35ae71d..9873944 100644 (file)
@@ -1,6 +1,6 @@
 
 /*
- * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
+ * Copyright (c) 2013-2014 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,
@@ -190,8 +190,9 @@ public class ICMP extends Packet {
             end += rawPayload.length;
         }
         int checksumStartByte = start + getfieldOffset(CHECKSUM) / NetUtils.NumBitsInAByte;
+        int even = end & ~1;
 
-        for (int i = start; i <= (end - 1); i = i + 2) {
+        for (int i = start; i < even; i = i + 2) {
             // Skip, if the current bytes are checkSum bytes
             if (i == checksumStartByte) {
                 continue;
@@ -199,7 +200,13 @@ public class ICMP extends Packet {
             wordData = ((data[i] << 8) & 0xFF00) + (data[i + 1] & 0xFF);
             sum = sum + wordData;
         }
-        carry = (sum >> 16) & 0xFF;
+        if (even < end) {
+            // Add the last octet with zero padding.
+            wordData = (data[even] << 8) & 0xFF00;
+            sum = sum + wordData;
+        }
+
+        carry = sum >>> 16;
         finalSum = (sum & 0xFFFF) + carry;
         return (short) ~((short) finalSum & 0xFFFF);
     }
index 3363f42..56793c4 100644 (file)
@@ -260,7 +260,17 @@ public class IPv4 extends Packet {
      */
     public void setHeaderField(String headerField, byte[] readValue) {
         if (headerField.equals(PROTOCOL)) {
-            payloadClass = protocolClassMap.get(readValue[0]);
+            // Don't set payloadClass if framgment offset is not zero.
+            byte[] fragoff = hdrFieldsMap.get(FRAGOFFSET);
+            if (fragoff == null || BitBufferHelper.getShort(fragoff) == 0) {
+                payloadClass = protocolClassMap.get(readValue[0]);
+            }
+        } else if (headerField.equals(FRAGOFFSET)) {
+            if (readValue != null && BitBufferHelper.getShort(readValue) != 0) {
+                // Clear payloadClass because protocol header is not present
+                // in this packet.
+                payloadClass = null;
+            }
         } else if (headerField.equals(OPTIONS) &&
                    (readValue == null || readValue.length == 0)) {
             hdrFieldsMap.remove(headerField);
index e81fbf0..287b73a 100644 (file)
@@ -1,6 +1,6 @@
 
 /*
- * Copyright (c) 2013 Cisco Systems, Inc. and others.  All rights reserved.
+ * Copyright (c) 2013-2014 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,
@@ -9,6 +9,8 @@
 
 package org.opendaylight.controller.sal.packet;
 
+import java.util.Arrays;
+
 import junit.framework.Assert;
 
 import org.junit.Test;
@@ -74,28 +76,58 @@ public class ICMPTest {
                 (byte) 0x2b, (byte) 0x2c, (byte) 0x2d, (byte) 0x2e,
                 (byte) 0x2f, (byte) 0x30, (byte) 0x31, (byte) 0x32,
                 (byte) 0x33, (byte) 0x34, (byte) 0x35, (byte) 0x36, (byte) 0x37 };
+        serializeTest(icmpRawPayload, (short)0xe553);
+
+        serializeTest(null, (short)0xb108);
+        serializeTest(new byte[0], (short)0xb108);
+
+        byte[] odd = {
+            (byte)0xba, (byte)0xd4, (byte)0xc7, (byte)0x53,
+            (byte)0xf8, (byte)0x59, (byte)0x68, (byte)0x77,
+            (byte)0xfd, (byte)0x27, (byte)0xe0, (byte)0x5b,
+            (byte)0xd0, (byte)0x2e, (byte)0x28, (byte)0x41,
+            (byte)0xa3, (byte)0x48, (byte)0x5d, (byte)0x2e,
+            (byte)0x7d, (byte)0x5b, (byte)0xd3, (byte)0x60,
+            (byte)0xb3, (byte)0x88, (byte)0x8d, (byte)0x0f,
+            (byte)0x1d, (byte)0x87, (byte)0x51, (byte)0x0f,
+            (byte)0x6a, (byte)0xff, (byte)0xf7, (byte)0xd4,
+            (byte)0x40, (byte)0x35, (byte)0x4e, (byte)0x01,
+            (byte)0x36,
+        };
+        serializeTest(odd, (short)0xd0ad);
+
+        // Large payload that causes 16-bit checksum overflow more than
+        // 255 times.
+        byte[] largeEven = new byte[1024];
+        Arrays.fill(largeEven, (byte)0xff);
+        serializeTest(largeEven, (short)0xb108);
+
+        byte[] largeOdd = new byte[1021];
+        Arrays.fill(largeOdd, (byte)0xff);
+        serializeTest(largeOdd, (short)0xb207);
+    }
 
-        short checksum = (short)0xe553;
-
-        // Create ICMP object
+    private void serializeTest(byte[] payload, short checksum)
+        throws PacketException {
         ICMP icmp = new ICMP();
-        icmp.setType((byte)8);
-        icmp.setCode((byte)0);
-        icmp.setIdentifier((short) 0x46f5);
-        icmp.setSequenceNumber((short) 2);
-        icmp.setRawPayload(icmpRawPayload);
-        //icmp.setChecksum(checksum);
+        icmp.setType((byte)8).setCode((byte)0).
+            setIdentifier((short)0x46f5).setSequenceNumber((short)2);
+        int payloadSize = 0;
+        if (payload != null) {
+            icmp.setRawPayload(payload);
+            payloadSize = payload.length;
+        }
 
         // Serialize
-        byte[] stream = icmp.serialize();
-        Assert.assertTrue(stream.length == 64);
+        byte[] data = icmp.serialize();
+        Assert.assertEquals(payloadSize + 8, data.length);
 
         // Deserialize
         ICMP icmpDes = new ICMP();
-        icmpDes.deserialize(stream, 0, stream.length);
+        icmpDes.deserialize(data, 0, data.length);
 
         Assert.assertFalse(icmpDes.isCorrupted());
-        Assert.assertTrue(icmpDes.getChecksum() == checksum);
-        Assert.assertTrue(icmp.equals(icmpDes));
+        Assert.assertEquals(checksum, icmpDes.getChecksum());
+        Assert.assertEquals(icmp, icmpDes);
     }
 }
index f529871..b983428 100644 (file)
@@ -12,9 +12,9 @@ import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.util.Arrays;
 
-import junit.framework.Assert;
-
+import org.junit.Assert;
 import org.junit.Test;
+
 import org.opendaylight.controller.sal.match.Match;
 import org.opendaylight.controller.sal.match.MatchType;
 import org.opendaylight.controller.sal.utils.EtherTypes;
@@ -481,4 +481,200 @@ public class IPv4Test {
         Assert.assertEquals(protocol, (byte) match.getField(MatchType.NW_PROTO).getValue());
         Assert.assertEquals(tos, (byte) match.getField(MatchType.NW_TOS).getValue());
     }
+
+    @Test
+    public void testFragment() throws Exception {
+        byte[] payload1 = new byte[0];
+        byte[] payload2 = {
+            (byte)0x61, (byte)0xd1, (byte)0x3d, (byte)0x51,
+            (byte)0x1b, (byte)0x75, (byte)0xa7, (byte)0x83,
+        };
+        byte[] payload3 = {
+            (byte)0xe7, (byte)0x0f, (byte)0x2d, (byte)0x7e,
+            (byte)0x15, (byte)0xba, (byte)0xe7, (byte)0x6d,
+            (byte)0xb5, (byte)0xc5, (byte)0xb5, (byte)0x37,
+            (byte)0x59, (byte)0xbc, (byte)0x91, (byte)0x43,
+            (byte)0xb5, (byte)0xb7, (byte)0xe4, (byte)0x28,
+            (byte)0xec, (byte)0x62, (byte)0x6b, (byte)0x6a,
+            (byte)0xd1, (byte)0xcb, (byte)0x79, (byte)0x1e,
+            (byte)0xfc, (byte)0x82, (byte)0xf5, (byte)0xb4,
+        };
+
+        // Ensure that the payload is not deserialized if the fragment offset
+        // is not zero.
+        byte proto = IPProtocols.TCP.byteValue();
+        fragmentTest(payload1, proto, (short)0xf250);
+        fragmentTest(payload2, proto, (short)0xf248);
+        fragmentTest(payload3, proto, (short)0xf230);
+
+        proto = IPProtocols.UDP.byteValue();
+        fragmentTest(payload1, proto, (short)0xf245);
+        fragmentTest(payload2, proto, (short)0xf23d);
+        fragmentTest(payload3, proto, (short)0xf225);
+
+        proto = IPProtocols.ICMP.byteValue();
+        fragmentTest(payload1, proto, (short)0xf255);
+        fragmentTest(payload2, proto, (short)0xf24d);
+        fragmentTest(payload3, proto, (short)0xf235);
+
+        // Ensure that the protocol header in the first fragment is
+        // deserialized.
+        proto = IPProtocols.TCP.byteValue();
+        TCP tcp = new TCP();
+        tcp.setSourcePort((short)1234).setDestinationPort((short)32000).
+            setSequenceNumber((int)0xd541f5f8).setAckNumber((int)0x58da787d).
+            setDataOffset((byte)5).setReserved((byte)0).
+            setHeaderLenFlags((short)0x18).setWindowSize((short)0x40e8).
+            setUrgentPointer((short)0x15f7).setChecksum((short)0x0d4e);
+        firstFragmentTest(tcp, payload1, proto, (short)0xdfe6);
+        tcp.setChecksum((short)0xab2a);
+        firstFragmentTest(tcp, payload2, proto, (short)0xdfde);
+        tcp.setChecksum((short)0x1c75);
+        firstFragmentTest(tcp, payload3, proto, (short)0xdfc6);
+
+        proto = IPProtocols.UDP.byteValue();
+        UDP udp = new UDP();
+        udp.setSourcePort((short)53).setDestinationPort((short)45383).
+            setLength((short)(payload1.length + 8)).setChecksum((short)0);
+        firstFragmentTest(udp, payload1, proto, (short)0xdfe7);
+        udp.setLength((short)(payload2.length + 8));
+        firstFragmentTest(udp, payload2, proto, (short)0xdfdf);
+        udp.setLength((short)(payload3.length + 8));
+        firstFragmentTest(udp, payload3, proto, (short)0xdfc7);
+
+        proto = IPProtocols.ICMP.byteValue();
+        ICMP icmp = new ICMP();
+        icmp.setType((byte)8).setCode((byte)0).setIdentifier((short)0x3d1e).
+            setSequenceNumber((short)1);
+        firstFragmentTest(icmp, payload1, proto, (short)0xdff7);
+        firstFragmentTest(icmp, payload2, proto, (short)0xdfef);
+        firstFragmentTest(icmp, payload3, proto, (short)0xdfd7);
+    }
+
+    private void fragmentTest(byte[] payload, byte proto, short checksum)
+        throws Exception {
+        // Construct a fragmented raw IPv4 packet.
+        int ipv4Len = 20;
+        byte[] rawIp = new byte[ipv4Len + payload.length];
+
+        byte ipVersion = 4;
+        byte dscp = 35;
+        byte ecn = 2;
+        byte tos = (byte)((dscp << 2) | ecn);
+        short totalLen = (short)rawIp.length;
+        short id = 22143;
+        short offset = 0xb9;
+        byte ttl = 64;
+        byte[] srcIp = {(byte)0x0a, (byte)0x00, (byte)0x00, (byte)0x01};
+        byte[] dstIp = {(byte)0xc0, (byte)0xa9, (byte)0x66, (byte)0x23};
+
+        rawIp[0] = (byte)((ipVersion << 4) | (ipv4Len >> 2));
+        rawIp[1] = tos;
+        rawIp[2] = (byte)(totalLen >>> Byte.SIZE);
+        rawIp[3] = (byte)totalLen;
+        rawIp[4] = (byte)(id >>> Byte.SIZE);
+        rawIp[5] = (byte)id;
+        rawIp[6] = (byte)(offset >>> Byte.SIZE);
+        rawIp[7] = (byte)offset;
+        rawIp[8] = ttl;
+        rawIp[9] = proto;
+        rawIp[10] = (byte)(checksum >>> Byte.SIZE);
+        rawIp[11] = (byte)checksum;
+        System.arraycopy(srcIp, 0, rawIp, 12, srcIp.length);
+        System.arraycopy(dstIp, 0, rawIp, 16, srcIp.length);
+        System.arraycopy(payload, 0, rawIp, ipv4Len, payload.length);
+
+        // Deserialize.
+        IPv4 ipv4 = new IPv4();
+        ipv4.deserialize(rawIp, 0, rawIp.length * Byte.SIZE);
+
+        Assert.assertEquals(ipVersion, ipv4.getVersion());
+        Assert.assertEquals(ipv4Len, ipv4.getHeaderLen());
+        Assert.assertEquals(dscp, ipv4.getDiffServ());
+        Assert.assertEquals(ecn, ipv4.getECN());
+        Assert.assertEquals(totalLen, ipv4.getTotalLength());
+        Assert.assertEquals(id, ipv4.getIdentification());
+        Assert.assertEquals((byte)0, ipv4.getFlags());
+        Assert.assertEquals(offset, ipv4.getFragmentOffset());
+        Assert.assertEquals(ttl, ipv4.getTtl());
+        Assert.assertEquals(proto, ipv4.getProtocol());
+        Assert.assertEquals(checksum, ipv4.getChecksum());
+        Assert.assertEquals(NetUtils.byteArray4ToInt(srcIp),
+                            ipv4.getSourceAddress());
+        Assert.assertEquals(NetUtils.byteArray4ToInt(dstIp),
+                            ipv4.getDestinationAddress());
+        Assert.assertFalse(ipv4.isCorrupted());
+
+        // payloadClass should not be set if fragment offset is not zero.
+        Assert.assertEquals(null, ipv4.getPayload());
+        Assert.assertArrayEquals(payload, ipv4.getRawPayload());
+    }
+
+    private void firstFragmentTest(Packet payload, byte[] rawPayload,
+                                   byte proto, short checksum)
+        throws Exception {
+        // Construct a raw IPv4 packet with MF flag.
+        int ipv4Len = 20;
+        payload.setRawPayload(rawPayload);
+        byte[] payloadBytes = payload.serialize();
+        byte[] rawIp = new byte[ipv4Len + payloadBytes.length];
+
+        byte ipVersion = 4;
+        byte dscp = 13;
+        byte ecn = 1;
+        byte tos = (byte)((dscp << 2) | ecn);
+        short totalLen = (short)rawIp.length;
+        short id = 19834;
+        byte flags = 0x1;
+        short offset = 0;
+        short off = (short)(((short)flags << 13) | offset);
+        byte ttl = 64;
+        byte[] srcIp = {(byte)0xac, (byte)0x23, (byte)0x5b, (byte)0xfd};
+        byte[] dstIp = {(byte)0xc0, (byte)0xa8, (byte)0x64, (byte)0x71};
+
+        rawIp[0] = (byte)((ipVersion << 4) | (ipv4Len >> 2));
+        rawIp[1] = tos;
+        rawIp[2] = (byte)(totalLen >>> Byte.SIZE);
+        rawIp[3] = (byte)totalLen;
+        rawIp[4] = (byte)(id >>> Byte.SIZE);
+        rawIp[5] = (byte)id;
+        rawIp[6] = (byte)(off >>> Byte.SIZE);
+        rawIp[7] = (byte)off;
+        rawIp[8] = ttl;
+        rawIp[9] = proto;
+        rawIp[10] = (byte)(checksum >>> Byte.SIZE);
+        rawIp[11] = (byte)checksum;
+        System.arraycopy(srcIp, 0, rawIp, 12, srcIp.length);
+        System.arraycopy(dstIp, 0, rawIp, 16, srcIp.length);
+        System.arraycopy(payloadBytes, 0, rawIp, ipv4Len, payloadBytes.length);
+
+        // Deserialize.
+        IPv4 ipv4 = new IPv4();
+        ipv4.deserialize(rawIp, 0, rawIp.length * Byte.SIZE);
+
+        Assert.assertEquals(ipVersion, ipv4.getVersion());
+        Assert.assertEquals(ipv4Len, ipv4.getHeaderLen());
+        Assert.assertEquals(dscp, ipv4.getDiffServ());
+        Assert.assertEquals(ecn, ipv4.getECN());
+        Assert.assertEquals(totalLen, ipv4.getTotalLength());
+        Assert.assertEquals(id, ipv4.getIdentification());
+        Assert.assertEquals(flags, ipv4.getFlags());
+        Assert.assertEquals(offset, ipv4.getFragmentOffset());
+        Assert.assertEquals(ttl, ipv4.getTtl());
+        Assert.assertEquals(proto, ipv4.getProtocol());
+        Assert.assertEquals(checksum, ipv4.getChecksum());
+        Assert.assertEquals(NetUtils.byteArray4ToInt(srcIp),
+                            ipv4.getSourceAddress());
+        Assert.assertEquals(NetUtils.byteArray4ToInt(dstIp),
+                            ipv4.getDestinationAddress());
+        Assert.assertFalse(ipv4.isCorrupted());
+
+        // Protocol header in the first fragment should be deserialized.
+        Assert.assertEquals(null, ipv4.getRawPayload());
+
+        Packet desPayload = ipv4.getPayload();
+        Assert.assertEquals(payload, desPayload);
+        Assert.assertFalse(desPayload.isCorrupted());
+        Assert.assertArrayEquals(rawPayload, desPayload.getRawPayload());
+    }
 }