Programming a CLI Plugin with Python in SR Linux

#srlinux #python #cliplugin

In this post, you’ll see how to write your Python code to build a custom command, mostly over a ‘show fabric’ example that we’ve seen in the previous post.

This post may require some basic Python knowledge.

I’ll probably refer to a lot to keep this short and less boring. Also not to repeat the same information you can find out there. 🙂

First, let’s talk about the Python-based CLI framework of SR Linux.

Python CLI Plugin Framework

The SR Linux allows users to extend three types of CLI command trees by adding custom commands, and they are;

  • Global: Commands that can be called from any level in the CLI hierarchy
  • Tools: Commands under the tools tree
  • Show: The show commands like ‘show fabric’

Once the custom commands are loaded, they work just like any other native CLI commands including the auto-completion.

Now, let’s start coding it!

Coding a Custom Show Command: ‘show fabric’

I’ll explain this part mostly over the fabric.py file which is the source code of the show fabric commands.

The very first thing is importing the necessary modules(CliPlugin, Syntax, etc.) but I’ll let you figure it out from the examples.

To begin with, we must define our plugin as a class.

class Plugin(CliPlugin):

Every CLI plugin must have the load() method that defines the syntax:

def load(self, cli, **_kwargs):
    fabric = cli.show_mode.add_command(Syntax('fabric', help='shows ...'))
    
    help = fabric.add_command(Syntax('help', help='requires ...')\
    ,update_location=False, callback=self._show_help)
    
    summary = fabric.add_command(Syntax('summary', help='shows ...')\
    ,update_location=False, callback=self._show_summary, schema=self._get_schema())
    
    platform = fabric.add_command(Syntax('platform', help='shows...')\
    ,update_location=False, callback=self._show_platform, schema=self._get_schema())

The load typically has 3 parameters; self, cli (references the CLI command tree), and **_kwargs. The cli object creates the fabric with cli.show_mode.add_command. Then the subcommands like help, summary, etc. are added to the fabric. And last, the call-back function and the schema are passed to the add_command().

Once we defined the syntax, we can simply follow the steps in the public documentation of Nokia which are:

But, I’ll mix the order of these steps. 🙂 Let’s start with the call-back function;

def _show_help (self,state,output,**_kwargs):
    print('''
    The 'show fabric' command shows you statistics and the status of the uplinks and BGP peerings. 
    Therefore it requires some inputs that need to be added in the 'fabric.py' file.
    
    '/etc/opt/srlinux/cli/plugins/fabric.py'
    
    Example:
    interfaces = []             # fill the subinterfaces list in or the description pattern
    description = "spine"
    uplink_peer_group = "eBGP-underlay"
    rr_peer_group = "to-vRR-overlay"
    ''')

A call-back function basically defines what you want from that show command. In this example, I just want it to print some text to help the user.

Let’s see another one that does more:

def _show_platform(self, state, output, **_kwargs):
    result_platform = Data(self._get_schema())
    self._set_formatters_platform(result_platform)
    with output.stream_data(result_platform):   
      self._populate_data_platform(result_platform, state)

As you see, the call-back function typically calls get_schema, set_formatters, populate_data, and finally prints it with the stream_data, which is all we need.

Now, let’s see what are those things. The schema:

def _get_schema(self):
    root = FixedSchemaRoot()
    platform_header = root.add_child(
        'platform_header',
        fields=['platform']
    )
    platform_header.add_child(
        'platform_child',
        key='Chassis Type',
        fields=['S/N', 'Uptime','CPU (%)','Memory (%)','Disk Partitions (%)']
    )
    uplink_header = root.add_child(
        'uplink_header',
        fields=['uplinks']
    )
    uplink_header.add_child(
        'uplink_child',
        key='Local Interface',
        fields=['Local Router', 'Link Status','eBGP Status','Remote Router', 'Remote Interface']
    )

It defines the headers and fields you plan to have in your show command, like this;

The next is the fetch method, which is primarily consumed by the populate method.

def _fetch_state_platform(self, state):
    chassis_path = build_path(f'/platform/chassis')
    self.chassis_data = state.server_data_store.stream_data(chassis_path, recursive=True)
    
    cpu_path = build_path(f'/platform/control[slot=*]/cpu[index=all]/total/average-5')
    self.cpu_data = state.server_data_store.stream_data(cpu_path, recursive=True)
    
    mem_path = build_path(f'/platform/control[slot=*]/memory/utilization')
    self.mem_data = state.server_data_store.stream_data(mem_path, recursive=True)

    disk_path = build_path(f'/platform/control[slot=*]/disk[name=*]/partition[name=*]/percent-used')
    self.disk_data = state.server_data_store.stream_data(disk_path, recursive=True)

Well, this one is rather easy to explain; it is basically where we define the YANG path of the data and then fetch it from the ‘state’ data store with either get_data() or in this case stream_data().

After that, we code the populate data method:

def _populate_data_platform(self, result, state):                   
    result.synchronizer.flush_fields(result)
    data = result.platform_header.create()
    server_data = self._fetch_state_platform(state)
    
    chassis_type = self.chassis_data.platform.get().chassis.get().type or '<Unknown>'
    data_child = data.platform_child.create(chassis_type)
    data_child.s_n = self.chassis_data.platform.get().chassis.get().serial_number or '<Unknown>'
    data_child.uptime = self._time_handler(state, self.chassis_data.platform.get().chassis.get().last_booted) or '<Unknown>'
    data_child.cpu____ = self.cpu_data.platform.get().control.get().cpu.get().total.get().average_5 or '<Unknown>'
    data_child.memory____ = self.mem_data.platform.get().control.get().memory.get().utilization or '<Unknown>'
    data_child.disk_partitions____ = self.disk_data.platform.get().control.get().disk.get().partition.get(d.name).percent_used

    data_child.synchronizer.flush_fields(data_child)
    result.synchronizer.flush_children(result.platform_header)  
    return result

The synchronizer.flush_fields() is simply used for streaming optimization. Then we call the header we defined in the schema method and fill the fields with the data we fetched before.

You may want to iterate if needed, like iterating to find some interfaces with the same attributes (MTU, description, VLAN, etc.) and list them.  🤷‍♂️

 Okay, now the last part:

def _set_formatters_header(self, data):
    data.set_formatter('/header',Border(TagValueFormatter()))  

def _set_formatters_platform(self, data):
    data.set_formatter('/platform_header/platform_child', ColumnFormatter())

def _set_formatters_stats(self, data):
    data.set_formatter('/stats_header/stats_child', ColumnFormatter())    

def _set_formatters_rr(self, data):
    data.set_formatter('/rr_header/rr_child', ColumnFormatter())

def _set_formatters_uplink(self, data):
    data.set_formatter('/uplink_header/uplink_child',\ 
    ColumnFormatter(horizontal_alignment={'Link Status':Alignment.Center}))

The formatters, basically the cosmetics… Whether you want to align, tag, add a border, indent, or whatever is possible with the formatters.

Here we are!

If you followed till here, congrats and thank you!

Hope you write your own and enjoy it!