eAPI Python script to look at ARP entries per VRF

I needed to see all the different ARP entries in each VRF, so I wrote up this little script to do just that. The ‘show vrf’ command in eAPI has not yet been converted to JSON, so I had to do some text parsing to get the VRF names, then use those names to grab the ARP entries. On line 4 you’ll see that I use the ‘text’ option for the output of the JSON reply. That allows me to run a command that hasn’t been converted yet and get the raw text output:

response = switch.runCmds( 1, ["show vrf"], "text" )

The output looks like this:

"output": "   Vrf         RD            Protocols       State         Interfaces \n----------- ------------- --------------- ---------------- ---------- \n   test        100:100       ipv4            no routing               \n   test2       101:101       ipv4            no routing               \n   test3       102:102       ipv4            no routing               \n\n"

Or in a more familiar format:

   Vrf         RD            Protocols       State         Interfaces
----------- ------------- --------------- ---------------- ----------
   test        100:100       ipv4            no routing
   test2       101:101       ipv4            no routing
   test3       102:102       ipv4            no routing

Then I take the output and use splitlines() to take each line (separated by newline) and insert them into a list:

lines = response[0]['output'].splitlines()

Now I iterate through each entry of the ‘show ip vrf’ output and issue a ‘show ip arp vrf’ with the VRF name. I use the range() function, starting at the 3rd line (since the first two are just header lines), and go through the end of the list. Then I use the split() method to split each line on whitespace, taking the first entry which corresponds to the VRF name. Finally, I can use that VRF name in my command.

for i in range(2, len(lines) - 1):
  vrfname = lines[i].split()[0]
  command = "show ip arp vrf " + vrfname

Here’s the script in its entirety:


eAPI script to try different IP addresses

I’ve been trying to use code to solve more problems around the lab lately, and thought I’d start posting some of the little scripts I write. Today I had plugged a device (device A) into a switch and didn’t know what the device had set for its gateway (I did have the IP of the device itself, and it wasn’t .1 or .254). I didn’t have access to the configuration, so I thought I’d write a script to go through possible IP addresses and see if one of them would take until the owner got back to me. I had another machine trying to ping the IP address from a different subnet, so if the right gateway address was configured on the Ethernet port, I should start getting pings:

Device A — Switch — Test pinging machine

Now I started my constant ping from to, then I created and ran this Python script to find the right address.

Arista JSON eAPI example

One of the great things about working with EOS is the ability to script with JSON-RPC.  No longer does a network admin need to do screen scraping, you can get clean, machine-friendly data from the switch using CLI syntax you’re familiar with.  I’ll outline a simple example using Python.

First add jsonrpclib to your Python environment:

sudo easy_install jsonrpclib

Now we can use that library to make scripting to EOS pretty easy:

from jsonrpclib import Server
switches = ["", "", ""]
username = "admin"
password = "admin"

So far I’ve setup a list of switch IP addresses, and a username/password to use to login to each of them. Now let’s do something useful:

# Going through all the switch IP addresses listed above
for switch in switches:
    urlString = "https://{}:{}@{}/command-api".format(username, password, switch) #1
    switchReq = Server( urlString ) #2
    # Display the current vlan list
    response = switchReq.runCmds( 1, ["show vlan"] ) #3
    print "Switch : " + switch + " VLANs: " 
    print response[0]["vlans"].keys() #4

Now I iterate through each of the switches in the list. On each iteration the script does the following:
1) Creates a string that defines the url to reach the API
2) Start creating a JSON-RPC request with the url
3) Finish building the JSON-RPC request and send the HTTP POST with the commands I want to run on the switch. The JSON response is stored in response. The JSON-RPC library returns the “result” field automatically, so there is no need to parse through the boilerplate JSON-RPC reply.
4) Print out each of the VLANs configured on the switch. The response from the switch is a list, so first I grab the first (in this case only) item indexed by 0. This gives me a dictionary. Next I use the vlans key to select an object from the dictionary. This returns another dictionary, which has the VLAN names as the keys (and details as the values). Since I want to print a list of all the VLANs, I use the keys() method which returns a list of all the keys in the dictionary. Here is the JSON that is being parsed:

   "jsonrpc": "2.0",
   "result": [
         "sourceDetail": "",
         "vlans": {
            "1": {
               "status": "active",
               "name": "default",
               "interfaces": {
                  "Ethernet14": {
                     "privatePromoted": false
                  "Ethernet15": {
                     "privatePromoted": false
                  "Ethernet16": {
                     "privatePromoted": false
                  "Ethernet17": {
                     "privatePromoted": false
                  "Ethernet13": {
                     "privatePromoted": false
               "dynamic": false
            "51": {
               "status": "active",
               "name": "VLAN0051",
               "interfaces": {
                  "Vxlan1": {
                     "privatePromoted": false
               "dynamic": false
            "61": {
               "status": "active",
               "name": "VLAN0061",
               "interfaces": {
                  "Vxlan1": {
                     "privatePromoted": false
               "dynamic": false
   "id": "CapiExplorer-123"

Here’s the full script that also adds a few lines to configure a vlan:

OVSDB Client in Python Part 2

In the last post I left off simply connecting to the OVSDB, and sending/receiving messages. Now I want to add in a couple things:
1) Handling of Echo messages – There should be periodic Echo messages sent back and forth between my client and the ovsdb-server
2) Handling of updates – When we issue a command like `monitor`, not only do we get a snapshot of the database at that time, but we also register ourselves for updates. So I’d like the client to keep listening for updates to OVSDB and allow me to see them

Using the `recv` socket command like we did in the beginning would require the client to keep looping around waiting for a message and that doesn’t seem very efficient. Instead I’m going to leverage select. What `select` will let me do is to wait for incoming data, then take action when we hear something, or take action when there is outgoing data waiting to be processed. Here’s an example:

def listen_for_messages(sock, message_queues):
    # To send something, add a message to queue and append sock to outputs
    inputs = [sock, sys.stdin]
    outputs = []
    while sock:
        readable, writable, exceptional = select(inputs, outputs, [])
        for s in readable:
            if s is sock:
                data = sock.recv(4096)
                print "recv:" + data
            elif s is sys.stdin:
                print sys.stdin.readline()
                print "error"
        for w in writable:
            if w is sock:
                print "error"

To start things off, I create a few lists that will hold the inputs and outputs I plan to monitor. In this function I initialize the inputs to be the socket connection and stdin, and outputs are empty. Next I create a while loop that will run as long as the socket is open: `while sock:`. Now we get to the select statement. I pass in the inputs and outputs that will be monitored, and the select will return whatever inputs or outputs have data that need to be processed: `readable, writable, exceptional = select(inputs, outputs, [])`
If there are any inputs, they will be assigned to the variable `readable` and we will enter the first for loop. There we check to see if there is socket data or stdin data. When we get data on a socket, we will simply copy that data into our message queue (this is a quick and dirty way to reply to an echo), and put the socket into the output list. If its something from stdin (i.e. user entering stuff on the keyboard), I will just close the socket for now. As you can imagine, one could easily use this to allow the user to send commands to the ovsdb-server with some extra modification.
Now I handle the output list, much the same way I did the input list. In this case, if there is socket data in the output list, I pull the data off the message queue, and send it out the socket. Then I remove the socket from the output queue to leave it empty again.
That should take care of our basic message handler. Using select, we’re able to handle various inputs and outputs in a efficient manner, and in this case, receive and send echo messages. This will keep the connection to the ovsdb-server alive for us. Also, if we had issued a monitor command, it will receive those updates from the server and print them to screen.

If another function wants to send messages out to the server, we would need to slightly modify things so that we can add messages to the queue and put the socket into the output list. I’ve posted a modified version of this code on github where you can have a look if interested: https://github.com/fredhsu/py-ovsdb-client

OVSDB client in Python

After reading Brent’s post on OVSDB, I was inspired to try my hand at interfacing with OVSDB via Python. I highly recommend reading his post to learn more about OVSDB before looking at the code examples below.

Making ovsdb-server listen

The first thing I needed to do was to make the database accept my client. If you’re running OVS version 1.10 or later, you can do this using ovs-appctl:

ovs-appctl -t ovsdb-server ovsdb-server/add-remote ptcp:6632

If you’ve installed OVS on Ubuntu using apt, and its already running, I found this worked for me:

  1. Edit /usr/share/openvswitch/scripts/ovs-ctl
  2. Under #Start ovsdb-server add a line:
  3. set "$@" --remote=ptcp:6632
  4. Then restart ovs
  5. sudo service openvswitch-switch restart

Now that we’ve got the server listening to us, we can start to tackle to code.

Python sockets

To connect to the DB, we’ll need to establish a TCP connection on the port we selected above. We’ll use Python sockets for this:

import socket


s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect((OVSDB_IP, OVSDB_PORT))

This sets up our socket connection, and from here on out we can send things over using s.send() and receive with s.recv()

As an easy test we can try running getting a list of the databases, which is done using JSON-RPC. For this case its pretty simple:

 "method :"list_dbs",
 "params": [],
 "id" : 0

First we send the query and then we get the response:

import json

list_dbs_query =  {"method":"list_dbs", "params":[], "id": 0}
response = s.recv(4096)
print response

The number passed to recv is the size of the buffer we want to use to receive. We take the response and put it in a variable which we can print out to see what databases are present. You should get something like this:


Now we’ve got two way communication going to the database. You’ll see we got an echo, the reply to our query, and another echo. As long as this stays open, we’ll still receive periodic echos from the server. To keep this connection open we’ll need to respond the echo messages we get from the server. I’ll tackle that in my next post.

ODL subnet REST API, Python, and some error handling

Its been a while since I’ve had a chance to play around on OpenDaylight for a while, so I thought I’d warm up with some Python and API calls. One thing I haven’t done much of with my code so far is handling errors, so in this post I’m going to access the Subnet API, and try to do some error handling as well from Python.

I recently found a nice HTTP client called Requests that simplifies the HTTP requests. You can add it using pip:

pip install requests

And then import it into your Python code:

import requests
from requests.auth import HTTPBasicauth

I’ve also imported the auth stuff for logging into the controller.
Now making a GET requests is simple, here is an example of querying the controller for all the configured subnets:

user = 'admin'
password = 'admin'
servierIP = ''
container = 'default'
allSubnets = '/controller/nb/v2/subnet/' + container + '/subnet/all'
url = 'http://' + serverIP + ':' + port + allSubnets
r = requests.get(url, auth=(user, password))
print r.json()

Requests can automatically take the response from the controller and give you back the JSON data. Now lets add some error handling by wrapping everything with try/except/else:

errorcodes = {
    400: 'Invalid data',
    401: 'User not authorized',
    409: 'Name conflict',
    404: 'Container name not found',
    500: 'Internal error',
    503: 'Service unavailable'

    r = requests.get(url, auth=(user, password))
except requests.exceptions.HTTPError as e:
    print e
    print "Reason : %s" % errorcodes[r.status_code]
    # No errors loading URL
    result = find_subnet(r.json()['subnetConfig'], subnetquery)
    print result

The first thing we do is use r.raise_for_status() which returns null if all goes well, otherwise it will raise an exception. One of the exceptions that can be raised is an HTTPError, in which case we’ll print out the error.

If everything is ok, then I pass the resulting JSON to a find_subnet function which just searches the JSON list for a particular subnet:

# given a list of subnets and a subnet to find, will return the subnet if found
# or None if not found
def find_subnet(subnets, subnetName):
    for subnet in subnets:
        if subnet['subnet'] == subnetName:
            return subnet
    return None

Nothing too crazy here, but just another Python example for those who are interested. Also, please note that the APIs have changed here and there, so be sure to check the URLs you are calling to make sure they are correct. I’ve found that the API docs on the OpenDaylight wiki are not always in sync with the version of controller I’m using. You can find the full script here:

Unit Testing OpenDaylight code with Mininet and Python

I recently got pinged by Dale Carder from the University of Wisconsin regarding a python API he is developing for ODL.  The API is a nice step in relieving some of the tedium when dealing with the ODL REST API.
One of the cool things he’s done as part of his project is create some unit tests for his code using the python API for mininet, coupled with his Python API code.  Unit tests are a great way to make sure the code you’re creating does what it should do, and keeps doing the right things when you make changes. They are key to practices such as Test Driven Development(TDD). Combining Mininet API calls with ODL API calls could be a powerful tool for creating network applications. Let’s take a closer look and see what he’s done to leverage that Mininet API:

First he creates a class for to define the topology:

class SingleSwitchTopo(Topo):
    "Single switch connected to n hosts."
    def __init__(self, n=2, **opts):
        # Initialize topology and default options
        Topo.__init__(self, **opts)
        # mininet/ovswitch does not want ':'s in the dpid
        switch_id = SWITCH_1.translate(None, ':')
        switch = self.addSwitch('s1', dpid=switch_id)
        for h in range(n):
            host = self.addHost('h%s' % (h + 1))
            self.addLink(host, switch)

This gives us a Mininet instance with a switch with hosts that can be used to test the API calls.

Next he starts up the test network:

def setup_mininet_simpleTest():
    "Create and test a simple network"
    topo = SingleSwitchTopo(n=4)
    #net = Mininet(topo)
    net = Mininet( topo=topo, controller=lambda name: RemoteController( 
                   name, ip=CONTROLLER ) )

Then finally he setups the tests and tests the API calls that he makes. Here is the setup and one of the test cases:

class TestSequenceFunctions(unittest.TestCase):
    """Tests for OpenDaylight

       At this point, tests for OpenDaylightFlow and OpenDaylightNode
       are intermingled.  These could be seperated out into seperate

    def setUp(self):
        odl = OpenDaylight()
        odl.setup['hostname'] = CONTROLLER
        odl.setup['username'] = USERNAME
        odl.setup['password'] = PASSWORD
        self.flow = OpenDaylightFlow(odl)
        self.node = OpenDaylightNode(odl)

        self.switch_id_1 = SWITCH_1

        self.odl_test_flow_1 = {u'actions': u'DROP',
           u'etherType': u'0x800',
           u'ingressPort': u'1',
           u'installInHw': u'true',
           u'name': u'odl-test-flow1',
           u'node': {u'@id': self.switch_id_1, u'@type': u'OF'},
           u'priority': u'500'}

        self.odl_test_flow_2 = {u'actions': u'DROP',
           u'etherType': u'0x800',
           u'ingressPort': u'2',
           u'installInHw': u'true',
           u'name': u'odl-test-flow2',
           u'node': {u'@id': self.switch_id_1, u'@type': u'OF'},
           u'priority': u'500'}

    def test_01_delete_flows(self):
        """Clean up from any previous test run, just delete these
            flows if they exist.


You can find the beginnings of his API here:

please keep in mind its still very early goings, and a work in progress. Thanks goes out to Dale for letting me borrow his code for this post.