Nornir: Tasks, Plugins and Templates

Nornir tasks are run against all or a subset of inventory members with the result formatted into a framework structured to show what was run against whom and the results. Tasks can be custom built python code or pre-built plugins that have been installed and imported.

Table Of Contents

Tasks

A task is a reusable piece of code that implements some functionality for a single host or group of hosts. It is input as an argument into a Python function and returns the result as data in a structured format that the framework can understand and print. The run method is used to execute the task feeding in any further optional arguments that may have been specified. print_result displays a user friendly output documenting what actions the tasks have performed against which hosts. This is not done in realtime, it is prettifying of a stored result.

from nornir import InitNornir
from nornir.core.task import Task, Result
from nornir_utils.plugins.functions import print_result

nr = InitNornir(config_file="config.yml")

def inventory_usernames(task):
    return Result(host=task.host, result=f"{task.host.name} username is {task.host.username}")
output = nr.run(task=inventory_usernames)

The returned data contains the following constructs:

• AggregatedResult: A dict-like object that aggregates the results of all devices. As each device is a dictionary the result for each host can be called by individually by referencing the key (hostname).
• MultiResult: A list-like object that gives access to the results of all sub-tasks for a device. The result of a task can be manually called using .result. If there are multiple task results the list index number of the task is also required.

print(output)
AggregatedResult (inventory_usernames): {'HME-ASR-WAN01': MultiResult: [Result: "inventory_usernames"], 'HME-SWI-VSS01': MultiResult: [Result: "inventory_usernames"]}
print(output['HME-SWI-VSS01'])
MultiResult: [Result: "inventory_usernames"]
print(output['HME-SWI-VSS01'].result)
'HME-SWI-VSS01 username is user1'
print(output['HME-SWI-VSS01'][0].result)
'HME-SWI-VSS01 username is user1'

print_result prettifies the output to show the name and result of each task as well as the inventory host it was run against.

Within the run method its possible to feed other arguments in as well as change the banner message from the task name to a custom message.

def test_input_task(task, text):
    return Result(host=task.host, result=f"{task.host.name} is {text}")
output = nr.run(name="my custom banner name", task=test_input_task, text="DOWN")

A single task (no function)

To run commands on a device the inventory needs to be initialised (InitNornir), a connection plugin task defined to perform the actions (napalm_get) and print_result used to return the result of the stored output of the task.

from nornir import InitNornir
from nornir_utils.plugins.functions import print_result
from nornir_napalm.plugins.tasks import napalm_get

nr = InitNornir(config_file="config.yml", dry_run=True)

results = nr.run(task=napalm_get, getters=["facts"])
print_result(results)

If the task is to only be run on a subset of hosts a filter is used to create a new inventory object to run the task against.

from nornir_napalm.plugins.tasks import napalm_cli

sw1 = nr.filter(site="london")
results = sw1.run(task=napalm_cli, commands=["show ip interface brief"])

Running configuration tasks is no different to running show command tasks except for the arguments fed into them. Unlike napalm and netmiko, scrapli uses different modules for single or multiple command application with the input being a string or list depending.

from nornir_scrapli.tasks import get_prompt, send_command, send_commands, send_configs

command_result = nr.run(task=send_command, command="show ip int brief")
command_results = nr.run(task=send_commands, commands=["show version", "show ip int brief"])

description = 'Configured by Scrapli through Nornir'
config_results = nr.run(task=send_configs, configs=["interface GigabitEthernet3", f"description { description }"])

Multiple sub-tasks in one task function

Grouping sub-tasks under a main task (function) gives a lot more control over the look and feel of the jobs been run as the script format and output can reflect the bigger picture (overall main task) rather than each component in it (each task).

from nornir import InitNornir
from nornir_utils.plugins.functions import print_result
from nornir_napalm.plugins.tasks import napalm_get, napalm_configure, napalm_cli

def multiple_tasks(task):
    task.run(task=napalm_cli, commands=["show ip ospf neighbor"])
    task.run(task=napalm_configure, dry_run=False, configuration="interface loopback 1000")
    task.run(task=napalm_get, getters=["interfaces"])

output = nr.run(task=multiple_tasks)
print_result(output)

Within AggregatedResult each host has a result for the main task (function) and one for each sub-task.

AggregatedResult (multiple_tasks): {
    'sw1': MultiResult: [
        Result: "multiple_tasks",
        Result: "napalm_cli",
        Result: "napalm_configure",
        Result: "napalm_get"],
    'sw2': MultiResult: [
        Result: "multiple_tasks",
        Result: "napalm_cli",
        Result: "napalm_configure",
        Result: "napalm_get"
        ]
    }

By default the main task (function called multiple_tasks) will be empty and each sub-task (napalm_cli, napalm_configure, napalm_get) has the data returned by the connection plugin.

Each task can be called and printed individually with the framework (print_result) or by printing the dictionary normally using .result. These two commands both print versions of the data returned by napalm_cli.

print_result(output['sw1'][1])
pprint(output['sw1'][1].result)

Changing the returned output

What is returned by a task is dictated by the plugin, it is normally the output of the show command or the configuration applied. When a task function is used to run a task an additional output can be returned for the function and the actual task output optionally suppressed.

The below example uses a function to gather the interfaces state (svi_interfaces) and uses the function output to return only the SVI interfaces. The parent function (main_task) runs this sub-task as well as another sub-task to gather ospf neighbors.

from nornir.core.task import Task, Result
from nornir_netmiko.tasks import netmiko_send_command
from nornir_utils.plugins.functions import print_result

def svi_interfaces(task):
    output = task.run(netmiko_send_command, command_string="show ip int brief")
    vlan_list = []
    for each_intf in output.result.splitlines():
        if 'Vlan' in each_intf:
            vlan_list.append(each_intf)
            vlan_intf = "\n".join(vlan_list)
    return Result(host=task.host, result=vlan_intf)

def main_task(task):
    task.run(netmiko_send_command, command_string="show ip ospf neighbor")
    task.run(svi_interfaces)

output = nr.run(task=main_task)

print_result(output) is the equivalent of print_result(output, vars=['result']), both produce the same output of all task results.

To filter out the result of the show ip int brief sub-task the severity level of this task must be changed from INFO to DEBUG. By default only tasks with a level of INFO will be printed out. dry_run can be set globally for the inventory at initialised or on a per-task basis in the run method. If set in both places the later tales precedence.

import logging

def svi_interfaces(task):
    output = task.run(netmiko_send_command, command_string="show ip int brief", severity_level=logging.DEBUG)
    vlan_list = []
    for each_intf in output.result.splitlines():
        if 'Vlan' in each_intf:
            vlan_list.append(each_intf)
            vlan_intf = "\n".join(vlan_list)
    return Result(host=task.host, result=vlan_intf)

An alternative take on this is for the function to return a custom variable instead using result. The limitation of doing it this way is that none of the other results can be printed as all tasks results are stored in result. You could try slicing the result, but using severity level is easier and neater.

def svi_interfaces(task):
    output = task.run(netmiko_send_command, command_string="show ip int brief")
    vlan_list = []
    for each_intf in output.result.splitlines():
        if 'Vlan' in each_intf:
            vlan_list.append(each_intf)
            vlan_intf = "\n".join(vlan_list)
    return Result(host=task.host, text="VLANs on this switch:", vlans=vlan_intf)

output = nr.run(task=main_task)
print_result(output, vars=['text', 'vlans'])

Finally as result is just another variable a custom message and the results could be printed by including all in the vars list.

print_result(output, vars=['text', 'vlans', 'result'])

Connection Plugins

Connection plugins are pre-built tasks that can be run on their own or nested within other tasks. Like any Nornir task they are a function with a task argument, optional other arguments (input commands or config files) and return Nornir structured data (AggregatedResult).

nornir_netmiko: netmiko_commit, netmiko_file_transfer, netmiko_save_config, netmiko_send_command, netmiko_send_config
nornir_napalm: napalm_cli, napalm_configure, napalm_get, napalm_ping, napalm_validate
nornir_scapli - Core: send_command.py, send_commands.py, send_commands_from_file.py, get_prompt.py, send_config.py, send_configs.py, send_configs_from_file.py, send_interactive.py
nornir_scapli – Config(Merge/replace configs): abort_config.py, commit_config.py, diff_config.py, get_config.py, get_version.py, load_config.py
nornir_scapli - Netconf: capabilities.py, commit.py, delete_config.py, discard.py, edit_config.py, get.py, get_config.py, lock.py, rpc.py, unlock.py, validate.py

To use any of these plugins the appropriate platform needs to be set in the inventory, the table below shows some of the options.

The proceeding single and multi-line configurations examples in this post are run on a C3560 switch so the defaults.yml file is defined as follows.

username: user1
password: my_password
connection_options:
  scrapli:
    platform: cisco_iosxe
  netmiko:
    platform: cisco_ios
  napalm:
    platform: ios

Before they can be used the plugins first to be installed and the relevant tasks imported.

pip install nornir-netmiko
pip install nornir-scrapli
pip install nornir_napalm

from nornir_netmiko.tasks import netmiko_send_config
from nornir_scrapli.tasks import send_config, send_configs
from nornir_napalm.plugins.tasks import napalm_configure

from nornir_utils.plugins.functions import print_result
from nornir import InitNornir

Netmiko

The configuration is applied using config_commands, it takes a string of one command or a list of multiple commands.

output = nr.run(netmiko_send_config, config_commands="interface loopback100")
output = nr.run(netmiko_send_config, config_commands=["interface loopback101","description Configured with nornir-netmiko"])

If using a variable of multi-line config (like from a rendered template) this will need converting into a list (split the new lines) before being fed into Netmiko. Alternatively this could be saved as a file and loaded using config_file instead of config_commands, I think this does something similar under the hood.

add_intf = "interface loopback102\n description Configured with nornir-netmiko"
output = nr.run(netmiko_send_config, config_commands=add_intf.split("\n"))

Netmiko does not support dry_run, if it is used the task will fail.

Scrapli

Scrapli uses separate tasks for applying single or multi config commands. Annoyingly the dictionary used for applying the configuration is different depending on the tasks, send_config uses config and send_configs uses configs.

output = nr.run(task=send_config, config="interface loopback200")
output = nr.run(task=send_configs, configs=["interface loopback201", "description Configured with nornir-scrapli"])

A variable containings multi-line config can either be input as a raw string with send_config or split into list of lines (like with Netmiko) with send_configs. send_configs_from_file allows a file to be used for the input configuration.

add_intf = "interface loopback202\n description Configured with nornir-scrapli"
output = nr.run(task=send_config, config=add_intf)
output = nr.run(task=send_configs, configs=add_intf.split("\n")

Scrapli supports dry_run but as it doesn’t return anything in the task result am not sure what use it is. The documentation says ‘Whether to apply changes or not; if dry run, will ensure that it is possible to enter config mode, but will NOT send any configs’. Can be applied to inventory or task.

nr = InitNornir(config_file="config.yml", dry_run=True)
output = nr.run(task=send_config, config="interface loopback202", dry_run=True)

Napalm

Naplam will merge or replace the configuration rather than applying the commands line-by-line. The the default action is merge, this can be changed using replace=True. The input can be a file (filename) or string (configuration), if it is multi-line config Napalm will automatically formatted it much like Scarpli send_configs does.

output = nr.run(task=napalm_configure, configuration="interface loopback300")
output = nr.run(task=napalm_configure, configuration="interface loopback301\n description Configured with nornir-napalm")

The new configuration is copied to the device (SCP), verified (verify /md5), differences checked (archive config incremental-diffs), a backup made of the running config (rollback_config.txt) and the configurations merged (copy flash:/merge_config.txt running-config).

%HA_EM-6-LOG: CLIlog: dir flash:/merge_config.txt
%HA_EM-6-LOG: CLIlog: verify /md5 flash:/merge_config.txt
%HA_EM-6-LOG: CLIlog: show archive config incremental-diffs flash:/merge_config.txt  ignorecase
%HA_EM-6-LOG: CLIlog: copy running-config flash:/rollback_config.txt
%HA_EM-6-LOG: CLIlog: copy flash:/merge_config.txt  running-config

Because Napalm first checks for differences it has the intelligence to know if changes are required. + or - show whether configuration is to be added or removed. If the same Nornir task was to be run a second time nothing would be returned as no changes are needed. Unlike Scrapli and Netmiko the resulting output is stored in the diff dictionary rather than result.

output = nr.run(task=napalm_configure, configuration="interface loopback300")
print(output['sw1'][0].result)
None
print(output['sw1'][0].diff)
'+interface loopback300'

Naplam fully supports dry_run using .diff to show the configuration that would be applied or removed if the task was to be run for real. dry_run can be set globally at inventory initialization or on a per-task basis in the run method. If set in both places the task level setting takes precedence.

nr = InitNornir(config_file="config.yml", dry_run=True)
output = nr.run(task=napalm_configure, configuration="interface loopback302", dry_run=True)

Nornir Templates

The nornir_jinja2 plugin renders jinja2 templates within the framework using either a file (template_file) or a string (template_string) as the template. The nornir_utils tasks load_yaml and write_file can be used to load the template file and save the rendered output to file.

• template_file: Looks in the specified path for the named template. Can optional use jinja filters.
• template_string: Uses the specified string as the template.
• load_yaml: Loads the contents of file (only takes the one argument of filename) which can then be saved as a host_var or group_var.
• write_file: Writes the contents locally to a filename. I will replace the contents of existing files unless append=True is set to append the contents to the file. The other optional setting of dry_run to dictate whether to apply the changes or not.

The following modules need installing foe the examples within this post.

from nornir_utils.plugins.tasks.data import load_yaml
from nornir_jinja2.plugins.tasks import template_file
from nornir_utils.plugins.tasks.files import write_file

from nornir_utils.plugins.functions import print_result
from nornir_napalm.plugins.tasks import napalm_configure
from nornir_scrapli.tasks import send_configs

from nornir import InitNornir
nr = InitNornir(config_file="config.yml")

import logging

Render one config snippet for all devices

This first example creates a generic config from the one input file input_vlans.yml to be used by all devices.

vlan:
  310: VL_310
  320: VL_320
  330: VL_330
ssh_acl:
  - { remark: Office Access }
  - { permit: 192.168.10.0 0.0.0.255 }
  - { remark: Citrix Access }
  - { permit: 172.16.10.0 0.0.0.255 }

As this only needs to be loaded and rendered once a filter is used to limit the inventory to just the first host. The result is assigned to a variable to be used by the template task, it could also be printed by the framework (print_result) or native Python.

nr_run_once = nr.filter(name=list(nr.inventory.hosts.keys())[0])
output = nr_run_once.run(task=load_yaml, file="input_data.yml")
vlan_acl_vars = output[list(output.keys())[0]].result

print_result(output)
print(vlan_acl_vars)

By default templates will use host_vars as the parent dictionary, they are references with {{ host.xxx }}. This example is not using host_vars to build VLAN and ACL (still used for hostname) so rather than host it uses an arbitrary name (input_vars) to match what is used in the template_file task.

hostname {{ host.name }}
!
{% for each_vlan, each_name in loaded_vars.vlan %}
vlan {{ each_vlan }}
 name {{ each_name }}
{% endfor %}
!
{% for each_vl in loaded_v_vars.vlan %}
vlan {{ each_vl.values() | list | first }}
 name {{ each_vl.keys() | list | first }}
{% endfor %}

Once the template has been rendered napalm_configure applies the configuration, netmiko_send_config is not used as does not support dry_run.

output = nr_run_once.run(task=template_file, template="vlan_acl_tmpl.j2", path="templates/", input_vars=vlan_acl_vars)
config = output[list(output.keys())[0]].result
task_output = nr.run(task=napalm_configure, dry_run=False, configuration=config))
print_result(task_output)

If the task fails due to a templating format error Nornir will put the host in a failed state and wont run any future task against it. Keep an eye out for this as is easy to miss when troubleshooting templating issues.
Either reset the failed host or set on_failed=True in the run.task. When testing templates it is easier to print the output at runtime.

print_result(nr_run_once.run(task=template_file, on_failed=True, template="vlan_acl_tmpl.j2", path="templates/", input_vars=vlan_acl_vars))
nr.data.failed_hosts
nr.data.reset_failed_hosts()

Ideally should be using a main task function containing sub-tasks as it gives a lot more flexibility. When running tasks individually (not as a task function) although they run concurrently host_vars and results have to be called by hostname (nr.inventory.hosts[‘sw1’].x or output[‘sw1’].result). When run as a sub-task the host is already selected so can they can be updated or called directly using task.host or config.result.

def generate_vlan_acl_config(task):
    file_output = task.run(task=load_yaml, file="input_data.yml")
    vlan_acl_vars = file_output.result
    template_output = task.run(task=template_file, template="vlan_acl_tmpl.j2", path="templates/", input_vars=vlan_acl_vars)
    task.host['config'] = template_output.result

nr_run_once = nr.filter(name=list(nr.inventory.hosts.keys())[0])
result = nr.run(task=generate_vlan_acl_config)

nr.inventory.hosts['sw2']['config']
'vlan 310\n name VL_310\nvlan 320\n name VL_320\nvlan 330\n name VL_330\n!\nip access-list standard SSH_ACCESS\n remark Office Access\n permit 192.168.10.0 0.0.0.255\n remark Citrix Access\n permit 172.16.10.0 0.0.0.255\n'

Render per-device config snippets

Nornir-template I would think has been designed more for this type of task as it allows for the rendering of a per-device config and assigning it as an inventory host_var. In this example a per-device SVI host_var (nr.inventory.hosts[‘sw1’].svi) is used to store the input data.

sw1:
  hostname: 10.10.10.1
  platform: ios
  data:
    svi:
      - { 310: 30.10.10.1 255.255.255.0 }
      - { 320: 30.10.20.1 255.255.255.0 }
      - { 330: 30.10.30.1 255.255.255.0 }
sw2:
  hostname: 10.10.10.1
  platform: ios
  data:
    svi:
      - { 310: 30.10.10.2 255.255.255.0 }
      - { 320: 30.10.20.2 255.255.255.0 }
      - { 330: 30.10.30.2 255.255.255.0 }

The template (svi_tmpl.j2) uses a host_var (host.svi) as the parent dictionary for the input variable used to render the template.

{% for each_svi in host.svi %}
interface {{ each_svi.keys() | list | first }}
 ip address {{ each_svi.values() | list | first }}
{% endfor %}

The template task is in its own separate function so that the task output can be suppressed (severity_level=logging.DEBUG) and a custom message returned (Result) in its place. Is no point printing the rendered config and the applied config as they are both the same.

def generate_config(task):
    config = task.run(task=template_file,
                      template="svi_tmpl.j2",
                      path="templates/",
                      severity_level=logging.DEBUG)
    task.host['config'] = config.result
    return Result(host=task.host, result = "SVI configurations generated")

The main function task calls the external template sub-task (generate_config) before running its own sub-task (napalm_configure) to apply the configuration. The configuration task has a custom banner (name=“Applying SVI config”) rather than the default banner of napalm_configure.

def create_svi(task):
    task.run(task=generate_config)
    task.run(task=napalm_configure, name="Applying SVI config", replace=False, dry_run=True, configuration=task.host["config"])

result = nr.run(task=create_svi)
print_result(result)

• ← Previous Post
• Next Post →