Netbox Check Device Sync Task

task api name: check_device_sync

The check_device_sync task performs a read-only sync-check against live devices and reports whether the data stored in NetBox is in sync with the actual device state. It does this by calling four existing sync sub-tasks in dry_run=True mode:

• interfaces — calls sync_device_interfaces(dry_run=True)
• mac_addresses — calls sync_mac_addresses(dry_run=True)
• ip_addresses — calls sync_device_ip(dry_run=True)
• bgp_peerings — calls sync_bgp_peerings(dry_run=True)

No data is written to NetBox. Each sub-check can be individually enabled or disabled.

Result Format

{
    # per-device summary — one entry per resolved device
    "result": {
        "ceos-spine-1": {
            "in_sync":       False,
            "interfaces":    True,
            "mac_addresses": False,
            "ip_addresses":  True,
            "bgp_peerings":  True,
        },
        ...
    },
    # per-category dry-run detail — full output from each sub-task
    "diff": {
        "interfaces":    { ... },   # dry-run result from sync_device_interfaces
        "mac_addresses": { ... },   # dry-run result from sync_mac_addresses
        "ip_addresses":  { ... },   # dry-run result from sync_device_ip
        "bgp_peerings":  { ... },   # dry-run result from sync_bgp_peerings
    },
}

A category is considered in sync when the corresponding dry-run reports no pending creates, updates, or deletes.

Sample Usage

NORFAB Netbox Check Device Sync Command Shell Reference

NorFab shell supports these command options for Netbox check_device_sync task:

nf# man tree netbox.check-sync.devices
root
└── netbox:    Netbox service
    └── check-sync:    Check if Netbox data is in sync with live device state
        └── devices:    Check if device data in NetBox is in sync with live device state
            ├── timeout:    Job timeout
            ├── workers:    Filter worker to target, default 'any'
            ├── verbose-result:    Control output details, default 'False'
            ├── progress:    Display progress events, default 'True'
            ├── instance:    Netbox instance name to target
            ├── branch:    Branching plugin branch name to use
            ├── devices:    List of NetBox devices to check sync state for
            ├── check-interfaces:    Check interfaces sync state, default 'True'
            ├── check-mac-addresses:    Check MAC addresses sync state, default 'True'
            ├── check-ip-addresses:    Check IP addresses sync state, default 'True'
            ├── check-bgp-peerings:    Check BGP peerings sync state, default 'True'
            ├── FO:    Filter hosts using Filter Object
            ├── 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
            ├── FX:    Filter hosts excluding them by name
            └── FN:    Negate the match
nf#

Python API Reference

Check if NetBox device data is in sync with live device data.

Calls sync_device_interfaces, sync_mac_addresses, sync_device_ip, and sync_bgp_peerings in dry-run mode and produces a per-device report indicating which items are in sync and which are not.

Result.diff contains the full dry-run detail from each sub-task, keyed by sub-task name (interfaces, mac_addresses, ip_addresses, bgp_peerings).

Parameters:

Name	Type	Description	Default
job	Job	NorFab Job object.	required
instance	str	NetBox instance name.	None
timeout	int	Timeout in seconds for Nornir jobs. Defaults to 60.	60
devices	list	List of device names to check.	None
branch	str	NetBox branching plugin branch name.	None
check_interfaces	bool	Check interface sync state. Defaults to True.	True
check_mac_addresses	bool	Check MAC address sync state. Defaults to True.	True
check_ip_addresses	bool	Check IP address sync state. Defaults to True.	True
check_bgp_peerings	bool	Check BGP peering sync state. Defaults to True.	True
**kwargs	Any	Nornir host filter arguments (e.g. FL, FC, FB).	{}

Returns:

Name	Type	Description
Result	Result	Per-device sync summary keyed by device name:: { "": { "in_sync": True | False, "interfaces": True | False, "mac_addresses": True | False, "ip_addresses": True | False, "bgp_peerings": True | False, } }

Source code in norfab\workers\netbox_worker\devices_tasks.py

@Task(
    fastapi={"methods": ["PATCH"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=CheckDeviceSyncInput,
    output=CheckDeviceSyncResult,
    mcp={
        "annotations": {
            "title": "Check Device Sync",
            "readOnlyHint": True,
            "destructiveHint": False,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def check_device_sync(
    self,
    job: Job,
    instance: Union[None, str] = None,
    timeout: int = 60,
    devices: Union[None, list] = None,
    branch: str = None,
    check_interfaces: bool = True,
    check_mac_addresses: bool = True,
    check_ip_addresses: bool = True,
    check_bgp_peerings: bool = True,
    **kwargs: Any,
) -> Result:
    """
    Check if NetBox device data is in sync with live device data.

    Calls ``sync_device_interfaces``, ``sync_mac_addresses``, ``sync_device_ip``,
    and ``sync_bgp_peerings`` in dry-run mode and produces a per-device report
    indicating which items are in sync and which are not.

    ``Result.diff`` contains the full dry-run detail from each sub-task, keyed by
    sub-task name (``interfaces``, ``mac_addresses``, ``ip_addresses``,
    ``bgp_peerings``).

    Args:
        job: NorFab Job object.
        instance (str, optional): NetBox instance name.
        timeout (int): Timeout in seconds for Nornir jobs. Defaults to 60.
        devices (list, optional): List of device names to check.
        branch (str, optional): NetBox branching plugin branch name.
        check_interfaces (bool): Check interface sync state. Defaults to True.
        check_mac_addresses (bool): Check MAC address sync state. Defaults to True.
        check_ip_addresses (bool): Check IP address sync state. Defaults to True.
        check_bgp_peerings (bool): Check BGP peering sync state. Defaults to True.
        **kwargs: Nornir host filter arguments (e.g. ``FL``, ``FC``, ``FB``).

    Returns:
        Result: Per-device sync summary keyed by device name::

            {
                "<device>": {
                    "in_sync": True | False,
                    "interfaces":    True | False,
                    "mac_addresses": True | False,
                    "ip_addresses":  True | False,
                    "bgp_peerings":  True | False,
                }
            }
    """
    devices = devices or []
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:check_device_sync",
        result={},
        resources=[instance],
        diff={},
    )

    # resolve devices from Nornir filters
    if kwargs:
        job.event("resolving devices from Nornir filters")
        nornir_hosts = self.get_nornir_hosts(kwargs, timeout)
        for host in nornir_hosts:
            if host not in devices:
                devices.append(host)
        job.event(
            f"resolved {len(nornir_hosts)} device(s) from Nornir filters, "
            f"{len(devices)} total device(s) selected"
        )

    if not devices:
        msg = "no devices specified"
        job.event(msg, severity="ERROR")
        ret.errors.append(msg)
        ret.failed = True
        return ret

    log.info(
        f"{self.name} - Check device sync for {len(devices)} device(s) in '{instance}'"
    )
    job.event(f"checking sync state for {len(devices)} device(s)")

    # initialize per-device result structure
    for device in devices:
        ret.result[device] = {}

    # --- check interfaces ---
    if check_interfaces:
        job.event("checking interfaces sync state")
        intf_result = self.sync_device_interfaces(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
        )
        if intf_result.errors:
            ret.errors.extend(intf_result.errors)
        for device, data in intf_result.result.items():
            in_sync = (
                not data.get("create")
                and not data.get("update")
                and not data.get("delete")
            )
            ret.result.setdefault(device, {})["interfaces"] = in_sync
        ret.diff["interfaces"] = intf_result.result

    # --- check MAC addresses ---
    if check_mac_addresses:
        job.event("checking MAC addresses sync state")
        mac_result = self.sync_mac_addresses(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
        )
        if mac_result.errors:
            ret.errors.extend(mac_result.errors)
        for device, data in mac_result.result.items():
            in_sync = not data.get("created") and not data.get("updated")
            ret.result.setdefault(device, {})["mac_addresses"] = in_sync
        ret.diff["mac_addresses"] = mac_result.result

    # --- check IP addresses ---
    if check_ip_addresses:
        job.event("checking IP addresses sync state")
        ip_result = self.sync_device_ip(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
        )
        if ip_result.errors:
            ret.errors.extend(ip_result.errors)
        for device, data in ip_result.result.items():
            in_sync = not data.get("created") and not data.get("updated")
            ret.result.setdefault(device, {})["ip_addresses"] = in_sync
        ret.diff["ip_addresses"] = ip_result.result

    # --- check BGP peerings ---
    if check_bgp_peerings:
        job.event("checking BGP peerings sync state")
        bgp_result = self.sync_bgp_peerings(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
        )
        if bgp_result.errors:
            ret.errors.extend(bgp_result.errors)
        for device, data in bgp_result.result.items():
            in_sync = (
                not data.get("create")
                and not data.get("update")
                and not data.get("delete")
            )
            ret.result.setdefault(device, {})["bgp_peerings"] = in_sync
        ret.diff["bgp_peerings"] = bgp_result.result

    checked_categories = {
        "interfaces": check_interfaces,
        "mac_addresses": check_mac_addresses,
        "ip_addresses": check_ip_addresses,
        "bgp_peerings": check_bgp_peerings,
    }
    for device, device_data in ret.result.items():
        device_data["in_sync"] = all(
            device_data.get(category) is True
            for category, checked in checked_categories.items()
            if checked
        )

    log.info(
        f"{self.name} - Check device sync complete for {len(ret.result)} device(s)"
    )
    return ret