Nornir Service "Task" Task

The Nornir Service "Task" Task is a versatile component of NorFab's Nornir service, designed to execute any arbitrary Nornir task plugin function. This task provides network automation and developer engineers with the flexibility to run custom Nornir tasks, enabling them to tailor their network automation workflows to meet specific requirements.

Key features of the Nornir Service "Task" Task include:

• Custom Task Execution: The "Task" Task allows you to run custom Nornir task functions, which can be referenced using the OS path to the custom task Python file stored on broker or using dot notation to reference an import module.

• Integration with Nornir Plugins: The Nornir framework supports a wide range of community-built plugins, which can be called directly or leveraged to extend the functionality of your custom tasks. By integrating these plugins, you can enhance your automation capabilities and streamline complex network operations. Reference the Nornir Plugins page for a list of available plugins.

• Scalability and Reusability: Custom Nornir tasks can be designed to be scalable and reusable, allowing you to apply the same task logic across different network environments and scenarios. This promotes consistency and efficiency in your network automation workflows, reducing the need for repetitive coding and manual intervention.

Nornir Tasks Sample Usage

Example of calling Nornir custom task function stored on NORFAB broker under nornir_tasks/echo.py file path:

├───inventory.yaml
└───nornir_tasks
    └───echo.py

Task echo.py takes provided arguments and echoes them back in results:

from nornir.core.task import Result, Task


def task(task: Task, **kwargs) -> Result:
    task.name = "echo"
    return Result(host=task.host, result=kwargs)

Example

CLIPython

C:\nf>nfcli
Welcome to NorFab Interactive Shell.
nf#
nf#nornir
nf[nornir]#task
nf[nornir-task]#plugin nf://nornir_tasks/echo.py arguments {"foo": "bar"} FC spine
ceos-spine-1:
    echo:
        foo: bar
ceos-spine-2:
    echo:
        foo: bar
nf[nornir-task]#top
nf#

Demo

Above runs echo.py custom Nornir task taking arguments {"foo": "bar"} as an input and echoing them back. Task only executed for Nornir hosts that contain ceos-spine in their hostname as we use FC - "Filter Contains" Nornir hosts targeting filter.

inventory.yaml should be located in same folder where we start nfcli, unless nfcli -i path_to_inventory.yaml flag used. Refer to Getting Started section on how to construct inventory.yaml file

This code is complete and can run as is

import pprint

from norfab.core.nfapi import NorFab

if __name__ == '__main__':
    nf = NorFab(inventory="inventory.yaml")
    nf.start()

    client = nf.make_client()

    res = client.run_job(
        service="nornir",
        task="task",
        kwargs={
            "plugin": "nf://nornir_tasks/echo.py",
            "argument": {"foo": "bar"},
            "FC": "ceos-spine"    
        }
    )

    pprint.pprint(res)

    nf.destroy()

Once executed, above code should produce this output:

C:\nf>python nornir_task_docs.py
{'nornir-worker-1': {'errors': [],
                    'failed': False,
                    'messages': [],
                    'result': {'ceos-spine-1': {'echo': {'argument': {'foo': 'bar'}}},
                                'ceos-spine-2': {'echo': {'argument': {'foo': 'bar'}}}},
                    'task': 'nornir-worker-1:task'}}

Refer to Getting Started section on how to construct inventory.yaml file.

Use Community Module Task

It is possible to run any Nornir task plugin created by open source community. For example, to use netmiko_send_commands from nornir_netmiko module need to set plugin argument to nornir_netmiko.tasks.netmiko_send_commands value and supply arguments option to provide further task parameters.

Example

CLIPython

C:\nf>nfcli
Welcome to NorFab Interactive Shell.
nf#
nf#nornir
nf[nornir]#task
nf[nornir-task]#plugin "nornir_netmiko.tasks.netmiko_send_command" arguments {"command_string": "show hostname"} FC spine
ceos-spine-1:
    netmiko_send_command:
        Hostname: ceos-spine-1
        FQDN:     ceos-spine-1
ceos-spine-2:
    netmiko_send_command:
        Hostname: ceos-spine-2
        FQDN:     ceos-spine-2
nf[nornir-task]#top
nf#

Demo

Above runs netmiko_send_command Nornir task from nornir_netmiko module and collects show hostname command output from hosts that contain ceos-spine in their host name since the use of targeting filter FC - "Filter Contains".

inventory.yaml should be located in same folder where we start nfcli, unless nfcli -i path_to_inventory.yaml flag used. Refer to Getting Started section on how to construct inventory.yaml file

This code is complete and can run as is

import pprint

from norfab.core.nfapi import NorFab

if __name__ == '__main__':
    nf = NorFab(inventory="inventory.yaml")
    nf.start()

    client = nf.make_client()

    res = client.run_job(
        service="nornir",
        task="task",
        kwargs={
            "plugin": "nornir_netmiko.tasks.netmiko_send_command",
            "command_string": "show hostname",
            "FC": "ceos-spine"    
        }
    )

    pprint.pprint(res)

    nf.destroy()

Notice slight difference, python api does not make use of arguments option and need to supply task parameters as is inside of kwargs dictionary.

Once executed, above code should produce this output:

C:\nf>python nornir_task_module_docs.py
{'nornir-worker-1': {'errors': [],
                    'failed': False,
                    'messages': [],
                    'result': {'ceos-spine-1': {'netmiko_send_command': 'Hostname: '
                                                                        'ceos-spine-1\n'
                                                                        'FQDN:     '
                                                                        'ceos-spine-1'},
                                'ceos-spine-2': {'netmiko_send_command': 'Hostname: '
                                                                        'ceos-spine-2\n'
                                                                        'FQDN:     '
                                                                        'ceos-spine-2'}}}

Refer to Getting Started section on how to construct inventory.yaml file.

NORFAB Nornir Task Shell Reference

NorFab shell supports these command options for Nornir task task:

nf#man tree nornir.task
root
└── nornir:    Nornir service
    └── task:    Run Nornir task
        ├── timeout:    Job timeout
        ├── workers:    Filter worker to target, default 'all'
        ├── add_details:    Add task details to results
        ├── run_num_workers:    RetryRunner number of threads for tasks execution
        ├── run_num_connectors:    RetryRunner number of threads for device connections
        ├── run_connect_retry:    RetryRunner number of connection attempts
        ├── run_task_retry:    RetryRunner number of attempts to run task
        ├── run_reconnect_on_fail:    RetryRunner perform reconnect to host on task failure
        ├── run_connect_check:    RetryRunner test TCP connection before opening actual connection
        ├── run_connect_timeout:    RetryRunner timeout in seconds to wait for test TCP connection to establish
        ├── run_creds_retry:    RetryRunner list of connection credentials and parameters to retry
        ├── tf:    File group name to save task results to on worker file system
        ├── tf_skip_failed:    Save results to file for failed tasks
        ├── diff:    File group name to run the diff for
        ├── diff_last:    File version number to diff, default is 1 (last)
        ├── progress:    Emit execution progress
        ├── table:    Table format (brief, terse, extend) or parameters or True
        ├── headers:    Table headers
        ├── headers_exclude:    Table headers to exclude
        ├── sortby:    Table header column to sort by
        ├── reverse:    Table reverse the sort by order
        ├── FB:    Filter hosts by name using Glob Patterns
        ├── FH:    Filter hosts by hostname
        ├── FC:    Filter hosts containment of pattern in name
        ├── FR:    Filter hosts by name using Regular Expressions
        ├── FG:    Filter hosts by group
        ├── FP:    Filter hosts by hostname using IP Prefix
        ├── FL:    Filter hosts by names list
        ├── FM:    Filter hosts by platform
        ├── FX:    Filter hosts excluding them by name
        ├── FN:    Negate the match
        ├── hosts:    Filter hosts to target
        ├── *plugin:    Nornir task.plugin.name to import or nf://path/to/plugin/file.py
        └── arguments:    Plugin arguments JSON formatted string
nf#

* - mandatory/required command argument

Python API Reference

Task to invoke any of supported Nornir task plugins. This function performs dynamic import of requested plugin function and executes nr.run using supplied args and kwargs

plugin attribute can refer to a file to fetch from file service. File must contain function named task accepting Nornir task object as a first positional argument, for example:

# define connection name for RetryRunner to properly detect it
CONNECTION_NAME = "netmiko"

# create task function
def task(nornir_task_object, **kwargs):
    pass

CONNECTION_NAME

CONNECTION_NAME must be defined within custom task function file if RetryRunner in use, otherwise connection retry logic skipped and connections to all hosts initiated simultaneously up to the number of num_workers.

Parameters:

Name	Type	Description	Default
plugin	str	(str) path.to.plugin.task_fun to import or nf://path/to/task.py to download custom task	required
kwargs		(dict) arguments to use with specified task plugin	{}

Source code in norfab\workers\nornir_worker.py

def task(self, plugin: str, **kwargs) -> Result:
    """
    Task to invoke any of supported Nornir task plugins. This function
    performs dynamic import of requested plugin function and executes
    ``nr.run`` using supplied args and kwargs

    ``plugin`` attribute can refer to a file to fetch from file service. File must contain
    function named ``task`` accepting Nornir task object as a first positional
    argument, for example:

    ```python
    # define connection name for RetryRunner to properly detect it
    CONNECTION_NAME = "netmiko"

    # create task function
    def task(nornir_task_object, **kwargs):
        pass
    ```

    !!! note "CONNECTION_NAME"

        ``CONNECTION_NAME`` must be defined within custom task function file if
        RetryRunner in use, otherwise connection retry logic skipped and connections
        to all hosts initiated simultaneously up to the number of ``num_workers``.

    :param plugin: (str) ``path.to.plugin.task_fun`` to import or ``nf://path/to/task.py``
        to download custom task
    :param kwargs: (dict) arguments to use with specified task plugin
    """
    # extract attributes
    add_details = kwargs.pop("add_details", False)  # ResultSerializer
    to_dict = kwargs.pop("to_dict", True)  # ResultSerializer
    filters = {k: kwargs.pop(k) for k in list(kwargs.keys()) if k in FFun_functions}
    ret = Result(task=f"{self.name}:task", result={} if to_dict else [])

    # download task from broker and load it
    if plugin.startswith("nf://"):
        function_text = self.fetch_file(plugin)
        if function_text is None:
            raise FileNotFoundError(
                f"{self.name} - '{plugin}' task plugin download failed"
            )

        # load task function running exec
        globals_dict = {}
        exec(function_text, globals_dict, globals_dict)
        task_function = globals_dict["task"]
    # import task function
    else:
        # below same as "from nornir.plugins.tasks import task_fun as task_function"
        task_fun = plugin.split(".")[-1]
        module = __import__(plugin, fromlist=[""])
        task_function = getattr(module, task_fun)

    self.nr.data.reset_failed_hosts()  # reset failed hosts
    filtered_nornir = FFun(self.nr, **filters)  # filter hosts

    # check if no hosts matched
    if not filtered_nornir.inventory.hosts:
        msg = (
            f"{self.name} - nothing to do, no hosts matched by filters '{filters}'"
        )
        log.debug(msg)
        ret.messages.append(msg)
        return ret

    nr = self._add_processors(filtered_nornir, kwargs)  # add processors

    # run task
    log.debug(f"{self.name} - running Nornir task '{plugin}', kwargs '{kwargs}'")
    with self.connections_lock:
        result = nr.run(task=task_function, **kwargs)
    ret.result = ResultSerializer(result, to_dict=to_dict, add_details=add_details)

    self.watchdog.connections_clean()

    return ret