BUG-3229: Add support for PacketIn filtering

When we are facing an overload situation, we may need to suppress
PacketIns while still processing other messages. This patch adds the
interface for ConnectionAdapter users which they can use to enable
filtering of such packets.

The implementation ensures that we filter these packets as efficiently
as possible.

Change-Id: I431631922437af7a08eb8ece647904d487634bd9
Signed-off-by: Robert Varga <rovarga@cisco.com>

diff --git a/openflow-protocol-api/src/main/java/org/opendaylight/openflowjava/protocol/api/connection/ConnectionAdapter.java b/openflow-protocol-api/src/main/java/org/opendaylight/openflowjava/protocol/api/connection/ConnectionAdapter.java
index 02679f05c7db8a2d3fc4027b8d146e98cc004e37..443dea5780d6d67b47f57f280bb3a55161a57892 100644 (file)
--- a/openflow-protocol-api/src/main/java/org/opendaylight/openflowjava/protocol/api/connection/ConnectionAdapter.java
+++ b/openflow-protocol-api/src/main/java/org/opendaylight/openflowjava/protocol/api/connection/ConnectionAdapter.java
@@ -82,4 +82,11 @@ public interface ConnectionAdapter extends OpenflowProtocolService {
     @Beta
     <T extends OutboundQueueHandler> OutboundQueueHandlerRegistration<T> registerOutboundQueueHandler(T handler,
         int maxQueueDepth, long maxBarrierNanos);
+
+    /**
+     * Set filtering of PacketIn messages. By default these messages are not filtered.
+     * @param enabled True if PacketIn messages should be filtered, false if they should be reported.
+     */
+    @Beta
+    void setPacketInFiltering(boolean enabled);
 }

diff --git a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/OFVersionDetector.java b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/OFVersionDetector.java
index 85bb5d55b8d6a2beacad65b91237d3fe86f0711e..083cf0eca8dc40d6dd3b73f8d779e41ab231886a 100644 (file)
--- a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/OFVersionDetector.java
+++ b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/OFVersionDetector.java
@@ -26,7 +26,9 @@ public class OFVersionDetector extends ByteToMessageDecoder {
     private static final byte OF10_VERSION_ID = EncodeConstants.OF10_VERSION_ID;
     /** Version number of OpenFlow 1.3 protocol */
     private static final byte OF13_VERSION_ID = EncodeConstants.OF13_VERSION_ID;
+    private static final short OF_PACKETIN = 10;
     private static final Logger LOGGER = LoggerFactory.getLogger(OFVersionDetector.class);
+    private volatile boolean filterPacketIns;
 
     /**
      * Constructor of class.
@@ -35,23 +37,31 @@ public class OFVersionDetector extends ByteToMessageDecoder {
         LOGGER.trace("Creating OFVersionDetector");
     }
 
+    public void setFilterPacketIns(final boolean enabled) {
+        filterPacketIns = enabled;
+    }
+
     @Override
-    protected void decode(final ChannelHandlerContext chc, final ByteBuf bb, final List<Object> list) throws Exception {
-        if (bb.readableBytes() == 0) {
+    protected void decode(final ChannelHandlerContext ctx, final ByteBuf in, final List<Object> out) {
+        if (!in.isReadable()) {
             LOGGER.debug("not enough data");
-            bb.release();
+            in.release();
             return;
         }
-        byte version = bb.readByte();
-        if ((version == OF13_VERSION_ID) || (version == OF10_VERSION_ID)) {
+
+        final byte version = in.readByte();
+        if (version == OF13_VERSION_ID || version == OF10_VERSION_ID) {
             LOGGER.debug("detected version: {}", version);
-            ByteBuf messageBuffer = bb.slice();
-            list.add(new VersionMessageWrapper(version, messageBuffer));
-            messageBuffer.retain();
+            if (!filterPacketIns || OF_PACKETIN != in.getUnsignedByte(in.readerIndex())) {
+                ByteBuf messageBuffer = in.slice();
+                out.add(new VersionMessageWrapper(version, messageBuffer));
+                messageBuffer.retain();
+            } else {
+                LOGGER.debug("dropped packetin");
+            }
         } else {
             LOGGER.warn("detected version: {} - currently not supported", version);
         }
-        bb.skipBytes(bb.readableBytes());
+        in.skipBytes(in.readableBytes());
     }
-
 }

diff --git a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/ConnectionAdapterImpl.java b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/ConnectionAdapterImpl.java
index 91f4ceb1f9dd4854c95b4f2b34c359f16c019731..464527c7b886ecef1da21bd61652dd5d7d36c778 100644 (file)
--- a/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/ConnectionAdapterImpl.java
+++ b/openflow-protocol-impl/src/main/java/org/opendaylight/openflowjava/protocol/impl/core/connection/ConnectionAdapterImpl.java
@@ -27,6 +27,8 @@ import java.util.concurrent.TimeUnit;
 import org.opendaylight.openflowjava.protocol.api.connection.ConnectionReadyListener;
 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandler;
 import org.opendaylight.openflowjava.protocol.api.connection.OutboundQueueHandlerRegistration;
+import org.opendaylight.openflowjava.protocol.impl.core.OFVersionDetector;
+import org.opendaylight.openflowjava.protocol.impl.core.PipelineHandlers;
 import org.opendaylight.openflowjava.statistics.CounterEventTypes;
 import org.opendaylight.openflowjava.statistics.StatisticsCounters;
 import org.opendaylight.yang.gen.v1.urn.opendaylight.openflow.protocol.rev130731.BarrierInput;
@@ -117,6 +119,7 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
     private OutboundQueueManager<?> outputManager;
     private boolean disconnectOccured = false;
     private final StatisticsCounters statisticsCounters;
+    private OFVersionDetector versionDetector;
     private final InetSocketAddress address;
 
     /**
@@ -135,6 +138,7 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
         this.address = address;
         channel.pipeline().addLast(output);
         statisticsCounters = StatisticsCounters.getInstance();
+
         LOG.debug("ConnectionAdapter created");
     }
 
@@ -455,6 +459,9 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
 
     @Override
     public void fireConnectionReadyNotification() {
+        versionDetector = (OFVersionDetector) channel.pipeline().get(PipelineHandlers.OF_VERSION_DETECTOR.name());
+        Preconditions.checkState(versionDetector != null);
+
         new Thread(new Runnable() {
             @Override
             public void run() {
@@ -514,4 +521,10 @@ public class ConnectionAdapterImpl implements ConnectionFacade {
     Channel getChannel() {
         return channel;
     }
+
+    @Override
+    public void setPacketInFiltering(final boolean enabled) {
+        versionDetector.setFilterPacketIns(enabled);
+        LOG.debug("PacketIn filtering {}abled", enabled ? "en" : "dis");
+    }
 }