Python

Python Script in NSO

• Use NSO Python modules and connect to the NSO datastore from your script, using transactions to read and write data.

• Useful for onetime operations like importing device configurations or generating reports from the configurations.

• Using the Python code in service packages:

  • Implement complex calculations and integrations with external systems in the code.

  • Populate and apply XML templates from Python.

  • Get the benefits of the FASTMAP algorithm.

Using Python API for Scripting

• NSO Python library consist of

  • low-level APIs: direct mappin of C API

  • high-level APIs: abstraction of low-level APIs

Management Agent API (MAAPI))

• a module that allows for transactional access of configuration and operational data

import ncs
with ncs.maapi.single_write_trans('admin', 'system') as t:
  t.get_elem('/devices/device{internet-rtr0}/address')

Some Example

Print names and addreses NSO is using to manage devices

import ncs # Import ncs module that contains submodules to work with Python API
m = ncs.maapi.Maapi() # Initialize new Maapi object
m.start_user_session('admin', 'system', []) # Start new user session that allow for multiple transactions
trans = m.start_read_trans() # Start new read transaction, not possible to write data to CDB with it
root = ncs.maagic.get_root(trans) # Get access to the root of the CDB using Maagic API
for dev in root.devices.device: # Iterate through devices in NSO using Maagic root object
    print(f"Device {dev.name} address {dev.address}") # Print names and address NSO is using for management of the device
m.close() #Ends session and closes socket. 

Change hostname of device

import ncs
with ncs.maapi.single_write_trans('admin', 'system') as t:
    root = ncs.maagic.get_root(t)
    root.devices.device['core-rtr0'].config.hostname = 'test-device'
    t.apply()

Add device to NSO

cat >| ${LAB_DIR}/add_device.py << EOF
import argparse
import ncs

def parse_args():
    parser = argparse.ArgumentParser()
    parser.add_argument('--name', help='device name', required=True)
    parser.add_argument('--address', help='device address', required=True)
    parser.add_argument('--ned', help='device NED ID', required=True)
    parser.add_argument('--port', help='device port', type=int, default=22)
    parser.add_argument('--desc', help='device description', default='Device created by maagic_create_device.py')
    parser.add_argument('--auth', help='device authgroup', default='default')
    return parser.parse_args()

def main(args):
    with ncs.maapi.Maapi() as m:
        with ncs.maapi.Session(m, 'admin', 'system'):
            with m.start_write_trans() as t:
                print(f'Setting the device "{args.name}" configuration...')

                # Get a reference to the device list
                root = ncs.maagic.get_root(t)
                device_list = root.devices.device

                if args.name not in device_list:
                    device = device_list.create(args.name)
                    device.address = args.address
                    device.port = args.port
                    device.description = args.desc
                    device.authgroup = args.auth
                    dev_type = device.device_type.cli
                    dev_type.ned_id = args.ned
                    device.state.admin_state = 'unlocked'
                    print('Committing the device configuration...')
                    t.apply()
                    print('Device committed!')
                else:
                    print(f'Device "{args.name}" configuration already exists...')

            # This transaction is no longer valid - since we are moving
            # back under ncs.maapi.Session(m, 'admin', 'python')

            # fetch-host-keys and sync-from does not require a
            # transaction, continue using the Maapi object
            root = ncs.maagic.get_root(m)
            device = root.devices.device[args.name]
            print('Fetching SSH keys...')
            output = device.ssh.fetch_host_keys()
            print(f'Result: {output.result}')
            print('Syncing configuration...')
            output = device.sync_from()
            print(f'Result: {output.result}')
            if not output.result:
                print(f'Error: {output.info}')


if __name__ == '__main__':
    main(parse_args())
EOF

Datastore Navigation (MAAGIC)

• enable Python dot notation access to the datastore data

# Access list item
router = root.devices.device['internet-rtr0']

# Create list item
root.services.l3vpn.create('test-l3vpn')

# Check if list item with a key exists
'internet-rtr0' in root.devices.device

# Set leaf value
root.devices.device['internet-rtr0'].address = '10.0.0.1'

# Remove list item
del root.devices.device['internet-rtr0']

# Note: Special characters are replaced with underscores (for example, element my-address becomes my_address).

FASTMAP Service

• ncs.application module is for developing Python-based NSO applications like services.

Using Python in Service Package

• Use ncs-make-package command to create a new Python and template-based service:

  • ncs-make-package --service-skeleton python-and-template --component-class l2vpn.L2vpn l2vpn

  • Note: It is not mandatory to provide the --component-class parameter to the ncs-make-package command. Without it, the module name is main.py. It is good practice to provide a meaningful module name for easier searching through log entries that the Python code generates and also for easier navigation between files during development.

  • see l2vpn/package-meta-data.xml for more information

  • Below is the image explaining the mapping of service in python code

Example

Creating a Python and Template-based service

• ncs-make-package --no-java --service-skeleton python-and-template --component-class loopback.Loopback --dest ${NCS_RUN_DIR}/packages/loopback loopback

• edit yang file as following:

cat >| ${NCS_RUN_DIR}/packages/loopback/src/yang/loopback.yang << EOF
module loopback {
namespace "http://com/example/loopback";
prefix loopback;

import tailf-ncs {
  prefix ncs;
}

import ietf-inet-types {
  prefix inet;
}

import tailf-common {
  prefix tailf;
}

list loopback {
  key name;

  uses ncs:service-data;
  ncs:servicepoint loopback-servicepoint;

  leaf name {
    tailf:info "Service instance name";
    type string;
  }

  leaf device {
    tailf:info "Device name";
    type leafref {
      path "/ncs:devices/ncs:device/ncs:name";
    }
  }

  leaf loopback-intf {
    tailf:info "Loopback interface number";
    type uint32;
  }

  leaf ip-prefix {
    tailf:info "IP prefix used to configure an address loopback interface";
    type inet:ipv4-prefix;
    mandatory true;
  }
}
}
EOF

• Building the package: make -C ${NCS_RUN_DIR}/packages/loopback/src/

• Create Python Mapping Logic by editting ${NCS_DIR}/packages/loopback/python/loopback/loopback.py

• cb_create function From:

# -*- mode: python; python-indent: 4 -*-
import ncs
from ncs.application import Service


# ------------------------
# SERVICE CALLBACK EXAMPLE
# ------------------------
class ServiceCallbacks(Service):

    # The create() callback is invoked inside NCS FASTMAP and
    # must always exist.
    @Service.create
    def cb_create(self, tctx, root, service, proplist):
        self.log.info('Service create(service=', service._path, ')')

        vars = ncs.template.Variables()
        vars.add('DUMMY', '127.0.0.1')
        template = ncs.template.Template(service)
        template.apply('loopback-template', vars)
    ...

# ---------------------------------------------
# COMPONENT THREAD THAT WILL BE STARTED BY NCS.
# ---------------------------------------------
class Loopback(ncs.application.Application):
    def setup(self):
        # The application class sets up logging for us. It is accessible
        # through 'self.log' and is a ncs.log.Log instance.
        self.log.info('Loopback RUNNING')

        # Service callbacks require a registration for a 'service point',
        # as specified in the corresponding data model.
        #
        self.register_service('loopback-servicepoint', ServiceCallbacks)

• To:

import ncs
import ipaddress
from ncs.application import Service

class ServiceCallbacks(Service):
    @Service.create
    def cb_create(self, tctx, root, service, proplist):
        self.log.info('Service create(service=', service._path, ')')

        ip_prefix = service.ip_prefix
        self.log.debug(f'Value of ip-prefix leaf is {ip_prefix}')
        net = ipaddress.IPv4Network(ip_prefix)
        ip_address = next(net.hosts())

Create and apply a service template

• Use NSO CLI to Create a Service Template

ncs_cli -C -u admin
config
devices device dist-rtr0 config interface Loopback 100
ip address 192.168.10.20 255.255.255.255
top
devices device core-rtr0 config interface Loopback 100
ipv4 address 192.168.10.20 255.255.255.255
top
commit dry-run outformat xml
end no-confirm
exit

• Edit the template file: ${NCS_RUN_DIR}/packages/loopback/templates/loopback-template.xml

• Or can use below command to edit it inline:

cat >| ${NCS_RUN_DIR}/packages/loopback/templates/loopback-template.xml << EOF
<config-template xmlns="http://tail-f.com/ns/config/1.0">
  <devices xmlns="http://tail-f.com/ns/ncs">
      <device>
         <name>{/device}</name>
         <config>             
           <interface xmlns="http://tail-f.com/ned/cisco-ios-xr">
             <Loopback>
               <id>{/loopback-intf}</id>
               <ipv4>
                 <address>
                   <ip>{\$IP_ADDRESS}</ip>
                   <mask>255.255.255.255</mask>
                 </address>
               </ipv4>
             </Loopback>
           </interface>
           <interface xmlns="urn:ios">
             <Loopback>
               <name>{/loopback-intf}</name>
               <ip>
                 <address>
                   <primary>
                     <address>{\$IP_ADDRESS}</address>
                     <mask>255.255.255.255</mask>
                   </primary>
                 </address>
               </ip>
             </Loopback>
           </interface>             
         </config>
       </device>
  </devices>
</config-template>
EOF

• Populate and Apply a Service Template from Python by editting ${NCS_RUN_DIR}/packages/loopback/python/loopback/loopback.py and add the following lines under the IP address calculation portion:

import ncs
import ipaddress
from ncs.application import Service

class ServiceCallbacks(Service):
    @Service.create
    def cb_create(self, tctx, root, service, proplist):
        self.log.info('Service create(service=', service._path, ')')

        ip_prefix = service.ip_prefix
        self.log.debug(f'Value of ip-prefix leaf is {ip_prefix}')
        net = ipaddress.IPv4Network(ip_prefix)
        ip_address = next(net.hosts())

        vars = ncs.template.Variables()
        vars.add('IP_ADDRESS', ip_address)
        template = ncs.template.Template(service)
        template.apply('loopback-template', vars)

• Or edittin the file inline as following:

cat >| ${NCS_RUN_DIR}/packages/loopback/python/loopback/loopback.py << EOF
# -*- mode: python; python-indent: 4 -*-
import ncs
import ipaddress
from ncs.application import Service


# ------------------------
# SERVICE CALLBACK EXAMPLE
# ------------------------
class ServiceCallbacks(Service):

    # The create() callback is invoked inside NCS FASTMAP and
    # must always exist.
    @Service.create
    def cb_create(self, tctx, root, service, proplist):
        self.log.info('Service create(service=', service._path, ')')

        ip_prefix = service.ip_prefix
        self.log.debug(f'Value of ip-prefix leaf is {ip_prefix}')
        net = ipaddress.IPv4Network(ip_prefix)
        ip_address = next(net.hosts())

        vars = ncs.template.Variables()
        vars.add('DUMMY', '127.0.0.1')
        vars.add('IP_ADDRESS', ip_address)
        template = ncs.template.Template(service)
        template.apply('loopback-template', vars) 


# ---------------------------------------------
# COMPONENT THREAD THAT WILL BE STARTED BY NCS.
# ---------------------------------------------
class Loopback(ncs.application.Application):
    def setup(self):
        # The application class sets up logging for us. It is accessible
        # through 'self.log' and is a ncs.log.Log instance.
        self.log.info('Loopback RUNNING')

        # Service callbacks require a registration for a 'service point',
        # as specified in the corresponding data model.
        #
        self.register_service('loopback-servicepoint', ServiceCallbacks)


    def teardown(self):
        # When the application is finished (which would happen if NCS went
        # down, packages were reloaded or some error occurred) this teardown
        # method will be called.

        self.log.info('Loopback FINISHED')
EOF

• The following image describe how service provisionin of Python and template services works

  1. You commit a new service instance with loopback interface number, device, and IP prefix.

  2. NSO invokes the Python create_cb() method, parameter IP prefix is available through the service parameter and calculation of usable IP address is done.

  3. The template is populated—IP address from Python code and loopback ID and device directly from the YANG model.

  4. Device configuration is created and pushed to the selected device using NED.

Troubleshooting a Python-based Service

Python Logs

• There are two types of logs in NSO

  • ncs-python-vm.log: Each NSO package starts in its own Python VM. This log collects the output from these VMs. The log also collects errors in Python code that occur when the packages reload or redeploy and prevent registration.

  • ncs-python-vm-loopback.log: The package-specific log collects log messages that you usually write by using the logging module in the package (in our case service) code.

• Connect to the NSO CLI and set the log level-debug so you can see the log message that Python code will write:

ncs_cli -C -u admin
config
python-vm logging level level-debug
commit
end

• Since only the Python code was changed, it is enough to use the packages package loopback redeploy command to update the running code. Then provision a new service instance, which generates a new message in the log:

packages package loopback redeploy
config
loopback test device core-rtr0 loopback-intf 202 ip-prefix 192.168.20.0/24
commit
end
exit

• Open and study the Python log for the loopback service. Find the log in the ${NCS_RUN_DIR}/log/ncs-python-vm-loopback.log file. There should be a line Value of ip-prefix leaf is 192.168.20.0/24:

tail -10 ${NCS_RUN_DIR}/logs/ncs-python-vm-loopback.log
<INFO> 29-Jan-2024::12:23:33.75 loopback ncs-dp-1427-loopback:loopback:0-1-th-509: - Service create(service=/loopback:loopback{test})
<DEBUG> 29-Jan-2024::12:23:33.76 loopback ncs-dp-1427-loopback:loopback:0-1-th-509: - Value of ip-prefix leaf is 192.168.21.0/24

Python API

Flow of NSO Maagic Python API

1. Create a transaction

• Common options:

  • single_read_trans

  • single_write_trans

with ncs.maapi.single_write_trans('admin', 'python', groups=['ncsadmin']) as t:
# groups setting is for system install as it is a Linux group

1. Access device information

import ncs
with ncs.maapi.single_read_trans('admin', 'python', groups=['ncsadmin']) as t:
    root = ncs.maagic.get_root(t)
    device_object  = root.devices.device["netsim-ios"]
    print(device_object.name)

• root.devices.device references all the devices in the NSO CDB. Even though the name says '.device' it is a YANG list, not one device.

• To determine available attributes, can reference the NSO YANG model, the NSO GUI, or use the print dir() option in Python

import ncs
with ncs.maapi.single_read_trans('admin', 'python', groups=['ncsadmin']) as t:
    root = ncs.maagic.get_root(t)
    device_object  = root.devices.device["netsim-ios"]
    print(dir(device_object))

1. Manipulate data

2. Apply configurations

3. Close the transaction

Manipulating Configuration

• Need to know namespace. This can be found by exploring NSO CLI (| display xpath) or using print(dir()) statement

admin@ncs# show running-config devices device netsim-ios config | display xpath
/devices/device[name='netsim-ios']/config/ios:tailfned/device netsim
/devices/device[name='netsim-ios']/config/ios:tailfned/police cirmode
/devices/device[name='netsim-ios']/config/ios:ip/source-route true
/devices/device[name='netsim-ios']/config/ios:ip/vrf[name='my-forward']/bgp/next-hop/Loopback 1
/devices/device[name='netsim-ios']/config/ios:ip/http/server false
/devices/device[name='netsim-ios']/config/ios:ip/http/secure-server false
/devices/device[name='netsim-ios']/config/ios:ip/community-list/number-standard[no='1']/permit
/devices/device[name='netsim-ios']/config/ios:ip/community-list/number-standard[no='2']/deny
/devices/device[name='netsim-ios']/config/ios:ip/community-list/standard[name='s']/permit
/devices/device[name='netsim-ios']/config/ios:class-map[name='a']/prematch match-all
/devices/device[name='netsim-ios']/config/ios:class-map[name='cmap1']/prematch match-all
/devices/device[name='netsim-ios']/config/ios:class-map[name='cmap1']/match/mpls/experimental/topmost [ 1 ]
/devices/device[name='netsim-ios']/config/ios:class-map[name='cmap1']/match/packet/length/max 255
/devices/device[name='netsim-ios']/config/ios:class-map[name='cmap1']/match/packet/length/min 2
/devices/device[name='netsim-ios']/config/ios:class-map[name='cmap1']/match/qos-group [ 1 ]
/devices/device[name='netsim-ios']/config/ios:policy-map[name='a']
/devices/device[name='netsim-ios']/config/ios:policy-map[name='map1']/class[name='c1']/drop
/devices/device[name='netsim-ios']/config/ios:policy-map[name='map1']/class[name='c1']/estimate/bandwidth
/devices/device[name='netsim-ios']/config/ios:policy-map[name='map1']/class[name='c1']/estimate/bandwidth/delay-one-in/doi 500
/devices/device[name='netsim-ios']/config/ios:policy-map[name='map1']/class[name='c1']/estimate/bandwidth/delay-one-in/milliseconds 100
/devices/device[name='netsim-ios']/config/ios:policy-map[name='map1']/class[name='c1']/priority
/devices/device[name='netsim-ios']/config/ios:policy-map[name='map1']/class[name='c1']/priority/percent 33
/devices/device[name='netsim-ios']/config/ios:interface/Loopback[name='0']
/devices/device[name='netsim-ios']/config/ios:interface/FastEthernet[name='0/0']
/devices/device[name='netsim-ios']/config/ios:interface/FastEthernet[name='1/0']
/devices/device[name='netsim-ios']/config/ios:interface/FastEthernet[name='1/1']
/devices/device[name='netsim-ios']/config/ios:spanning-tree/optimize/bpdu/transmission false
/devices/device[name='netsim-ios']/config/ios:router/bgp[as-no='64512']/aggregate-address/address 10.10.10.1
/devices/device[name='netsim-ios']/config/ios:router/bgp[as-no='64512']/aggregate-address/mask 255.255.255.251

Enable CDP

import ncs
with ncs.maapi.single_write_trans('admin', 'python', groups=['ncsadmin']) as t:
        root = ncs.maagic.get_root(t)
        device_object = root.devices.device["netsim-ios"].config
        device_object.ios__cdp.run = True
        t.apply()

# Result
admin@ncs# show running-config devices device netsim-ios config | display xpath
/devices/device[name='netsim-ios']/config/ios:tailfned/device netsim
/devices/device[name='netsim-ios']/config/ios:tailfned/police cirmode
/devices/device[name='netsim-ios']/config/ios:cdp/run true

Adding or Removin a Loopback

# Adding
import ncs
with ncs.maapi.single_write_trans('admin', 'python', groups=['ncsadmin']) as t:
    root = ncs.maagic.get_root(t)
    device_cdb = root.devices.device["netsim-ios"]
    device_cdb.config.ios__interface["Loopback"].create("1337")
    device_cdb.config.ios__interface.Loopback["1337"].ip.address.primary.address = "192.168.1.1"
    device_cdb.config.ios__interface.Loopback["1337"].ip.address.primary.mask = "255.255.255.252"
    t.apply()

# Removing
import ncs
with ncs.maapi.single_write_trans('admin', 'python', groups=['ncsadmin']) as t:
    root = ncs.maagic.get_root(t)
    device_cdb = root.devices.device["netsim-ios"]
    del device_cdb.config.ios__interface.Loopback["1337"]
    t.apply()

Examples of Python Script using NSO Python API

Check CDP

import ncs
with ncs.maapi.single_read_trans("admin", "python", groups=["ncsadmin"]) as t:
    root = ncs.maagic.get_root(t)
    device = root.devices.device["netsim-ios"]
    cdp_result = device.config.ios__cdp.run
    print(
        "For Device {}, CDP being enabled is {}".format(device.name, cdp_result)
    )

Check Interface

import ncs
with ncs.maapi.single_read_trans("admin", "python", groups=["ncsadmin"]) as t:
    root = ncs.maagic.get_root(t)
    device = root.devices.device["netsim-ios"]
    for interface in device.config.ios__interface:
        for if_type in device.config.ios__interface[interface]:
            if hasattr(if_type, "name"):
                print(
                    f'Device {device.name}, Interface {if_type} {if_type.name}'
                )

Reference

• NSO Python API

• ncs_pycli

• NSO Basics for Network Automation

• NSO Intermediate and Advanced Topics

• Basic Automation with Python

• Developing a Simple Service

• Applications in NSO