Containerlab Service Nornir Inventory Task

task api name: get_nornir_inventory

The Containerlab service get_nornir_inventory task is designed to generate a Nornir-compatible inventory for a specified lab. This task inspects the container lab environment and maps container details to Nornir inventory format, enabling seamless integration with network automation workflows.

Containerlab Nornir Inventory Task Overview

The get_nornir_inventory task provides the following features:

• Inventory Generation: Creates a Nornir-compatible inventory for a specified lab or all labs.
• Platform Mapping: Maps containerlab node kinds to Netmiko SSH platform types.
• Default Credentials: Optionally includes default credentials for containerlab nodes.

Containerlab Nornir Inventory Task Sample Usage

Below is an example of how to use the Containerlab get_nornir_inventory task to generate an inventory.

Example

DemoCLIPython

nf#containerlab get-nornir-inventory lab-name three-routers-lab
--------------------------------------------- Job Events -----------------------------------------------
05-May-2025 21:14:09.594 fc13d3b11ad54c2fb2330af20a7ceacd job started
05-May-2025 21:14:09.926 fc13d3b11ad54c2fb2330af20a7ceacd job completed in 0.332 seconds

--------------------------------------------- Job Results --------------------------------------------

containerlab-worker-1:
    ----------
    hosts:
        ----------
        r2:
            ----------
            hostname:
                192.168.1.130
            port:
                12203
            groups:
            platform:
                arista_eos
            username:
                admin
            password:
                admin
        r3:
            ----------
            hostname:
                192.168.1.130
            port:
                12204
            groups:
            platform:
            username:
                admin
            password:
                admin
        r1:
            ----------
            hostname:
                192.168.1.130
            port:
                12202
            groups:
            platform:
                arista_eos
            username:
                admin
            password:
                admin
nf#

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="containerlab",
        task="get_nornir_inventory",
        kwargs={
            "lab_name": "three-routers-lab",
        }
    )

    pprint.pprint(res)

    nf.destroy()

NORFAB Containerlab CLI Shell Reference

Below are the commands supported by the get_nornir_inventory task:

nf#man tree containerlab.get-nornir-inventory
root
└── containerlab:    Containerlab service
    └── get-nornir-inventory:    Get nornir inventory for a given lab
        ├── timeout:    Job timeout
        ├── workers:    Filter worker to target, default 'all'
        ├── verbose-result:    Control output details, default 'False'
        ├── lab-name:    Lab name to get Nornir inventory for
        ├── progress:    Display progress events, default 'True'
        └── groups:    List of groups to include in host's inventory
nf#

* - mandatory/required command argument

Python API Reference

Retrieves the Nornir inventory for a specified lab.

This method inspects the container lab environment and generates a Nornir-compatible inventory of hosts based on the lab's configuration. It maps containerlab node kinds to Netmiko SSH platform types and extracts relevant connection details.

Parameters:

Name	Type	Description	Default
lab_name	str	The name of the container lab to inspect. If not given loads inventory for all labs.	None
timeout	int	The timeout value for the inspection operation. Defaults to None.	None
groups	list	A list of group names to assign to the hosts in the inventory. Defaults to None.	None
use_default_credentials	bool	Use Containerlab default credentials for hosts or not.	True

Returns:

Name	Type	Description
Result	Result	A Result object containing the Nornir inventory. The result attribute
	Result	includes a dictionary with host details. If the lab is not found or an error occurs,
	Result	the failed attribute is set to True, and the errors attribute contains error messages.

Notes

• The method uses a predefined mapping (norfab.utils.platform_map) to translate containerlab node kinds to Netmiko platform types.
• If a container's SSH port cannot be determined, it is skipped, and an error is logged.
• The primary host IP address is determined dynamically using a UDP socket connection or by checking the host IP address in the container's port configuration.

Example of returned inventory structure

{ "hosts": { "host_name": { "hostname": "host_ip", "platform": "netmiko_platform", "groups": ["group1", "group2"], }, ...

Source code in norfab\workers\containerlab_worker\containerlab_worker.py

@Task(
    input=GetNornirInventoryInput,
    output=GetNornirInventoryResult,
    fastapi={"methods": ["GET"]},
    mcp={
        "annotations": {
            "title": "Get Nornir Inventory",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def get_nornir_inventory(
    self,
    job: Job,
    lab_name: Union[None, str] = None,
    timeout: int = None,
    groups: Union[None, list] = None,
    use_default_credentials: bool = True,
) -> Result:
    """
    Retrieves the Nornir inventory for a specified lab.

    This method inspects the container lab environment and generates a Nornir-compatible
    inventory of hosts based on the lab's configuration. It maps containerlab node kinds
    to Netmiko SSH platform types and extracts relevant connection details.

    Args:
        lab_name (str): The name of the container lab to inspect. If not given loads inventory for all labs.
        timeout (int, optional): The timeout value for the inspection operation. Defaults to None.
        groups (list, optional): A list of group names to assign to the hosts in the inventory. Defaults to None.
        use_default_credentials (bool, optional): Use Containerlab default credentials for hosts or not.

    Returns:
        Result: A `Result` object containing the Nornir inventory. The `result` attribute
        includes a dictionary with host details. If the lab is not found or an error occurs,
        the `failed` attribute is set to True, and the `errors` attribute contains error messages.

    Notes:
        - The method uses a predefined mapping (`norfab.utils.platform_map`) to translate containerlab
          node kinds to Netmiko platform types.
        - If a container's SSH port cannot be determined, it is skipped, and an error is logged.
        - The primary host IP address is determined dynamically using a UDP socket connection or
          by checking the host IP address in the container's port configuration.

    Example of returned inventory structure:
        {
            "hosts": {
                "host_name": {
                    "hostname": "host_ip",
                    "platform": "netmiko_platform",
                    "groups": ["group1", "group2"],
                },
                ...
    """
    timeout = timeout or 600
    groups = groups or []
    ret = Result(task=f"{self.name}:get_nornir_inventory", result={"hosts": {}})
    log.info(
        f"{self.name} - Get Nornir inventory: Building inventory for '{lab_name or 'all'}' lab(s)"
    )
    job.event(f"building nornir inventory for '{lab_name or 'all'}' lab(s)")

    # get lab details
    inspect = self.inspect(
        job=job, lab_name=lab_name, timeout=timeout, details=True
    )

    # return empty result if lab not found
    if not inspect.result:
        ret.failed = True
        ret.errors = [f"'{lab_name}' lab not found"]
        return ret

    # get host primary IP address
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    s.connect(("1.2.3.4", 12345))
    primary_host_ip = s.getsockname()[0]
    log.debug(
        f"{self.name} - determined Containerlab host primary IP - '{primary_host_ip}'"
    )

    # form hosts inventory
    for lname, containers in inspect.result.items():
        if lab_name and lname != lab_name:
            continue
        for container in containers:
            host_name = container["Labels"]["clab-node-name"]
            host_port = None
            host_ip = None

            # get ssh port
            for port in container["Ports"]:
                host_ip = primary_host_ip
                if port["port"] == 22 and port["protocol"] == "tcp":
                    host_port = port["host_port"]
                    # get host ip address
                    if port["host_ip"] not in [
                        "0.0.0.0",
                        "127.0.0.1",
                        "localhost",
                        "::",
                    ]:
                        host_ip = port["host_ip"]
                    break
            else:
                log.error(f"{self.name} - {host_name} Failed to map SSH port.")
                continue

            # add host to Nornir inventory
            ret.result["hosts"][host_name] = {
                "hostname": host_ip,
                "port": host_port,
                "groups": groups,
            }

            # get netmiko platform
            clab_platform_name = container["Labels"]["clab-node-kind"]
            netmiko_platform = PlatformMap.convert(
                "containerlab", "netmiko", clab_platform_name
            )
            if netmiko_platform:
                ret.result["hosts"][host_name]["platform"] = netmiko_platform[
                    "platform"
                ]
            else:
                log.warning(
                    f"{self.name} - {host_name} Clab-node-kind '{clab_platform_name}' not mapped to Netmiko platform."
                )
                continue

            # get default credentials
            if use_default_credentials:
                clab_platform = PlatformMap.get("containerlab", clab_platform_name)
                if not clab_platform:
                    log.warning(
                        f"{self.name} - {host_name} Clab-node-kind '{clab_platform_name}' not found."
                    )
                    continue
                if clab_platform.get("username"):
                    ret.result["hosts"][host_name]["username"] = clab_platform[
                        "username"
                    ]
                if clab_platform.get("password"):
                    ret.result["hosts"][host_name]["password"] = clab_platform[
                        "password"
                    ]

    return ret