Netbox Update Device Facts Task

task api name: update_device_facts

Limitations

Datasource nornir uses NAPALM get_facts getter and as such only supports these device platforms:

• Arista EOS
• Cisco IOS
• Cisco IOSXR
• Cisco NXOS
• Juniper JUNOS

Update Device Facts Sample Usage

NORFAB Netbox Update Device Facts Command Shell Reference

NorFab shell supports these command options for Netbox update_device_facts task:

Python API Reference

Function to update device facts in Netbox using information provided by NAPALM get_facts getter:

• serial number

Parameters:

Name	Type	Description	Default
instance	str	Netbox instance name	None
dry_run	bool	return information that would be pushed to Netbox but do not push it	False
datasource	str	service name to use to retrieve devices' data, default is nornir parse task	'nornir'
timeout	int	seconds to wait before timeout data retrieval job	60
batch_size	int	number of devices to process at a time	10
kwargs		any additional arguments to send to service for device data retrieval	{}

Returns:

Type	Description
dict	dictionary keyed by device name with updated details

Source code in norfab\workers\netbox_worker.py

def update_device_facts(
    self,
    instance: str = None,
    dry_run: bool = False,
    datasource: str = "nornir",
    timeout: int = 60,
    devices: list = None,
    batch_size: int = 10,
    **kwargs,
) -> dict:
    """
    Function to update device facts in Netbox using information
    provided by NAPALM get_facts getter:

    - serial number

    :param instance: Netbox instance name
    :param dry_run: return information that would be pushed to Netbox but do not push it
    :param datasource: service name to use to retrieve devices' data, default is nornir parse task
    :param timeout: seconds to wait before timeout data retrieval job
    :param batch_size: number of devices to process at a time
    :param kwargs: any additional arguments to send to service for device data retrieval
    :returns: dictionary keyed by device name with updated details
    """
    result = {}
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(task=f"{self.name}:update_device_facts", result=result)
    nb = self._get_pynetbox(instance)
    kwargs["add_details"] = True

    if datasource == "nornir":
        for i in range(0, len(devices), batch_size):
            kwargs["FL"] = devices[i : i + batch_size]
            kwargs["getters"] = "get_facts"
            self.event(
                f"retrieving facts data for devices {', '.join(kwargs['FL'])}",
                resource=instance,
            )
            data = self.client.run_job(
                "nornir",
                "parse",
                kwargs=kwargs,
                workers="all",
                timeout=timeout,
            )
            for worker, results in data.items():
                if results["failed"]:
                    log.error(
                        f"{worker} get_facts failed, errors: {'; '.join(results['errors'])}"
                    )
                    continue
                for host, host_data in results["result"].items():
                    if host_data["napalm_get"]["failed"]:
                        log.error(
                            f"{host} facts update failed: '{host_data['napalm_get']['exception']}'"
                        )
                        self.event(
                            f"{host} facts update failed",
                            resource=instance,
                            status="failed",
                            severity="WARNING",
                        )
                        continue
                    nb_device = nb.dcim.devices.get(name=host)
                    if not nb_device:
                        raise Exception(f"'{host}' does not exist in Netbox")
                    facts = host_data["napalm_get"]["result"]["get_facts"]
                    # update serial number
                    nb_device.serial = facts["serial_number"]
                    if not dry_run:
                        nb_device.save()
                    result[host] = {
                        "update_device_facts_dry_run"
                        if dry_run
                        else "update_device_facts": {
                            "serial": facts["serial_number"],
                        }
                    }
                    self.event(f"{host} facts updated", resource=instance)
    else:
        raise UnsupportedServiceError(
            f"'{datasource}' datasource service not supported"
        )

    return ret