PacketStreamer Service

Introduction

Packetstreamer is a packet streaming service that can selectively stream Openflow packets exchanged between any switch and the controller to an observer. It consists of two functional interfaces: (1) a REST-based interface to define the characteristics of the OpenFlow messages that are of interest, referred to as filter;  and (2) a Thrift-based interface for streaming the filtered packets.

REST API

The filter is defined with a POST request to the REST API: "http://<controller>:8080/wm/core/packettrace/json". The input data is parameters that define characteristics of the OpenFlow Messages we are interested in. Floodlight comes with a MAC-based filter, as an example. The following is the format for the filter.

{'mac':<hostMac>, 'direction':<direction>, 'period':<period>, 'sessionId':<sessionid>}

where,

The REST API returns the sessionId, which can be used to receive packets from the streaming thrift server. The return data is in json format,

{'sessionId':<sessionid>}

Here is an example in python to create a streaming session for 1000 seconds and a function to terminate a session.

url = 'http://%s:8080/wm/core/packettrace/json' % controller
filter = {'mac':host, 'direction':'both', 'period':1000}
post_data = json.dumps(filter)
request = urllib2.Request(url, post_data, {'Content-Type':'application/json'})
response_text = None

try:
    response = urllib2.urlopen(request)
    response_text = response.read()
except Exception, e:
    # Floodlight may not be running, but we don't want that to be a fatal
    # error, so we just ignore the exception in that case.
    print "Exception:", e
    exit

if not response_text:
    print "Failed to start a packet trace session"
    sys.exit()

response_text = json.loads(response_text)

sessionId = None
if "sessionId" in response_text:
    sessionId = response_text["sessionId"]

def terminateTrace(sid):
    global controller

    filter = {SESSIONID:sid, 'period':-1}
    post_data = json.dumps(filter)
    url = 'http://%s:8080/wm/core/packettrace/json' % controller
    request = urllib2.Request(url, post_data, {'Content-Type':'application/json'})
    try:
        response = urllib2.urlopen(request)
        response_text = response.read()
    except Exception, e:
        # Floodlight may not be running, but we don't want that to be a fatal
        # error, so we just ignore the exception in that case.
        print "Exception:", e

Thrift-Based Streaming Service

The packet streaming service is brokered by a Thrift-based streaming server. The Thrift interfaces are listed below. The complete  Thrift interface can be found at src/main/thrift/packetstreamer.thrift.

service PacketStreamer {

   /**
    * Synchronous method to get packets for a given sessionid
    */
   list<binary> getPackets(1:string sessionid),

   /**
    * Synchronous method to publish a packet.
    * It ensure the order that the packets are pushed
    */
   i32 pushMessageSync(1:Message packet),

   /**
    * Asynchronous method to publish a packet.
    * Order is not guaranteed.
    */
   oneway void pushMessageAsync(1:Message packet)

   /**
    * Terminate a session
    */
   void terminateSession(1:string sessionid)
}

The floodlight build script generates java and python libraries for the Thrift service. Other language support may be easily added by adding the language option into the build script, setup.sh.

The REST API section describes the creation of streaming session. Once the sessionId is created, the thrift interface, getPackets(sessionId), can be used to receive OFPackets for the given session. terminateSession(sessionId) can be used to terminate a live session.

A python example is below

try:
    # Make socket
    transport = TSocket.TSocket('localhost', 9090)
    # Buffering is critical. Raw sockets are very slow
    transport = TTransport.TFramedTransport(transport)
    # Wrap in a protocol
    protocol = TBinaryProtocol.TBinaryProtocol(transport)
    # Create a client to use the protocol encoder
    client = PacketStreamer.Client(protocol)
    # Connect!
    transport.open()

    while 1:
        packets = client.getPackets(sessionId)
        for packet in packets:
            print "Packet: %s"% packet
            if "FilterTimeout" in packet:
                sys.exit()

except Thrift.TException, e:
    print '%s' % (e.message)
    terminateTrace(sessionId)

Example Client

The current version of floodlight comes with a mac-based packet streaming example. Some pieces of the client code are listed in previous sections.
A complete python client example. which illustrates the usage of both the REST API and thrift client, can be found in floodlight source at net.floodlightcontroller.packetstreamer.
Make sure to install thrift on the client machine and give the correct path for packetstreamer's gen-py and thrift python directories in the example.

How to Extend to Service

The packet streaming service can be extended beyond the mac-based streaming example. net/floodlightcontroller/core/web/PacketTraceResource.java defines the REST interface. Matching field can be extended by adding fields in FilterParameters class. The matching logic is implemented in net/floodlightcontroller/core/OFMessageFilterManager.java and can be easily extended by adding matching logic in getMatchedFilters() method. A plugin framework could replace the current implementation if more use cases pop up.

Packet Format

The mac-based streaming example have the floodlight controller perform the packet formatting. The packet formatting is done in net/floodlightcontroller/core/OFMessageFilterManager.java. In some application, raw packet may be more useful. Format can be added to the packet-trace filter as part of the session properties. Based on the session property, the corresponding formatter could be called.