Netbox Check Device Sync Task

task api name: check_device_sync

Checks whether NetBox data is in sync with live device state without writing to NetBox. The task runs selected sync tasks in dry_run=True mode and returns a per-device summary plus detailed dry-run diffs.

How It Works

check_device_sync can run these read-only sub-checks:

• inventory — calls sync_device_inventory(dry_run=True)
• interfaces — calls sync_device_interfaces(dry_run=True)
• vrfs — calls sync_vrfs(dry_run=True) for VRFs, missing route targets and routing policies, and interface assignments
• vlans — calls sync_vlans(dry_run=True) for VLANs and interface assignments
• prefixes — calls sync_device_prefixes(dry_run=True)
• ip_addresses — calls sync_device_ip(dry_run=True)
• bgp_peerings — calls sync_bgp_peerings(dry_run=True)
• bgp_communities — calls sync_bgp_community(dry_run=True) for route targets and BGP plugin communities
• vrrp — calls sync_vrrp(dry_run=True)

Each sub-check can be enabled or disabled independently.

Sync Task Arguments

Use sync_kwargs to configure how individual dry-run synchronizers calculate drift. Supported keys are sync_device_inventory, sync_device_interfaces, sync_vrfs, sync_vlans, sync_device_prefixes, sync_device_ip, sync_vrrp, sync_bgp_community, and sync_bgp_peerings. Values are keyword argument dictionaries accepted by the corresponding task. Keep instance, timeout, devices, branch, and dry_run at the check_device_sync level; all subordinate tasks are always called with dry_run=True.

The mapping can be supplied inline or loaded from an nf:// YAML file:

sync_device_interfaces:
  filter_by_name: Loopback*
sync_vlans:
  filter_by_vlan_ids:
    - 100-299
sync_bgp_peerings:
  status: active

Output

Both coordinated tasks include inventory, prefixes, interfaces, VRFs, VLANs, IP addresses, VRRP, BGP communities, and BGP peerings. Every selected device has all nine category keys. Disabled categories are None; failed categories are False. Sync results contain child task summaries or dry-run previews, while checks return booleans for drift state. The diff dictionary retains child dry-run details under the same category keys, with None for unexecuted checks or stages. MAC synchronization is available through its standalone task.

{
    "result": {
        "ceos-spine-1": {
            "in_sync": False,
            "inventory": True,
            "interfaces": True,
            "vrfs": True,
            "vlans": False,
            "prefixes": True,
            "ip_addresses": True,
            "bgp_peerings": True,
            "bgp_communities": True,
            "vrrp": True,
        },
    },
    "diff": {
        "inventory": {},
        "interfaces": {},
        "vrfs": {},
        "vlans": {},
        "prefixes": {},
        "ip_addresses": {},
        "bgp_peerings": {},
        "bgp_communities": {},
        "vrrp": {},
    },
}

A category is considered in sync when the corresponding dry-run reports no pending creates, updates, or deletes. With ignore_deletions=True, pending deletions remain in Result.diff but do not make the category out of sync.

Notes / Gotchas

• No data is written to NetBox.
• Result.diff contains the raw dry-run detail from each enabled sub-task.
• Prefix, global VLAN, BGP community, and route-target drift affects every selected device; interface VLAN and VRRP drift is assessed per device.
• Device names can be supplied directly or resolved from Nornir filters.

Examples

CLIPython

Check all sync categories for explicit devices:

nf#netbox check-sync devices devices ceos-leaf-1 ceos-leaf-2

Check only interface and IP address sync:

nf#netbox check-sync devices devices ceos-leaf-1 check-inventory false check-vrfs false check-vlans false check-prefixes false check-bgp-peerings false check-bgp-communities false check-vrrp false

Resolve devices using a Nornir group filter:

nf#netbox check-sync devices FG leafs

Check against a NetBox branch:

nf#netbox check-sync devices devices ceos-leaf-1 branch my-branch

Load per-task arguments from the File Sharing service:

nf#netbox check-sync devices devices ceos-leaf-1 sync-kwargs nf://netbox/check_sync_kwargs.yaml

Context manager:

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()

    result = client.run_job(
        "netbox",
        "check_device_sync",
        workers="any",
        kwargs={
            "devices": ["ceos-leaf-1", "ceos-leaf-2"],
            "sync_kwargs": {
                "sync_device_interfaces": {"filter_by_name": "Loopback*"},
                "sync_vlans": {"filter_by_vlan_ids": ["100-299"]},
            },
        },
    )

Direct lifecycle:

from norfab.core.nfapi import NorFab

nf = NorFab(inventory="./inventory.yaml")
try:
    nf.start()
    client = nf.make_client()

    result = client.run_job(
        "netbox",
        "check_device_sync",
        workers="any",
        kwargs={
            "devices": ["ceos-leaf-1"],
            "check_inventory": False,
            "check_vrfs": False,
            "check_vlans": False,
            "check_prefixes": False,
            "check_bgp_peerings": False,
            "check_bgp_communities": False,
            "check_vrrp": False,
        },
    )

    filtered_result = client.run_job(
        "netbox",
        "check_device_sync",
        workers="any",
        kwargs={
            "FG": "leafs",
        },
    )
finally:
    nf.destroy()

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-inventory:    Check device inventory sync state, default 'True'
            ├── check-interfaces:    Check interfaces sync state, default 'True'
            ├── check-vrfs:    Check VRF and interface assignment sync state, default 'True'
            ├── check-vlans:    Check VLAN and interface assignment sync state, default 'True'
            ├── check-prefixes:    Check prefix sync state, default 'True'
            ├── check-ip-addresses:    Check IP addresses sync state, default 'True'
            ├── check-bgp-peerings:    Check BGP peerings sync state, default 'True'
            ├── check-bgp-communities:    Check BGP communities sync state, default 'True'
            ├── check-vrrp:    Check VRRP sync state, default 'True'
            ├── ignore-deletions:    Treat deletion-only differences as in sync, default 'False'
            ├── sync-kwargs:    Per-task sync arguments or nf:// YAML file
            ├── 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 the inventory, interface, VRF, VLAN, prefix, IP address, BGP peering, BGP community, and VRRP synchronizers in dry-run mode and produces a per-device report.

Result.diff contains the full dry-run detail from each sub-task, keyed by sub-task name (inventory, interfaces, vrfs, vlans, prefixes, ip_addresses, bgp_peerings, bgp_communities, vrrp).

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.	600
devices	list	List of device names to check.	None
branch	str	NetBox branching plugin branch name.	None
check_inventory	bool	Check device inventory sync state. Defaults to True.	True
check_interfaces	bool	Check interface sync state. Defaults to True.	True
check_vrfs	bool	Check VRF and interface assignment sync state.	True
check_vlans	bool	Check VLAN and interface assignment sync state.	True
check_prefixes	bool	Check prefix 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
check_bgp_communities	bool	Check BGP community sync state. Defaults to True.	True
check_vrrp	bool	Check VRRP sync state. Defaults to True.	True
ignore_deletions	bool	Treat deletion-only differences as in sync. Defaults to False.	False
sync_kwargs	Union[None, dict, str]	Per-task arguments keyed by sync task name, or an nf:// YAML file containing that dictionary. Supported keys are sync_device_inventory, sync_device_interfaces, sync_vrfs, sync_vlans, sync_device_prefixes, sync_device_ip, sync_vrrp, sync_bgp_community, and sync_bgp_peerings.	None
**kwargs	Any	Nornir host filter arguments (e.g. FL, FC, FB).	{}

Returns:

Name	Type	Description
Result	Result	Per-device summary with every category present. True means
	Result	no drift, False means drift or failure, and None means disabled.
	Result	Enabled checks continue after failures; errors and failed resources
	Result	are propagated. Diff contains each child dry-run diff, or None for
	Result	disabled checks. In_sync considers enabled checks only:: { "": { "in_sync": True | False, "inventory": True | False | None, "interfaces": True | False | None, "vrfs": True | False | None, "vlans": True | False | None, "prefixes": True | False | None, "ip_addresses": True | False | None, "bgp_peerings": True | False | None, "bgp_communities": True | False | None, "vrrp": True | False | None, } }

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

@Task(
    fastapi={"methods": ["POST"], "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 = 600,
    devices: Union[None, list] = None,
    branch: str = None,
    check_inventory: bool = True,
    check_interfaces: bool = True,
    check_vrfs: bool = True,
    check_vlans: bool = True,
    check_prefixes: bool = True,
    check_ip_addresses: bool = True,
    check_bgp_peerings: bool = True,
    check_bgp_communities: bool = True,
    check_vrrp: bool = True,
    ignore_deletions: bool = False,
    sync_kwargs: Union[None, dict, str] = None,
    **kwargs: Any,
) -> Result:
    """
    Check if NetBox device data is in sync with live device data.

    Calls the inventory, interface, VRF, VLAN, prefix, IP address, BGP
    peering, BGP community, and VRRP synchronizers in dry-run mode and produces
    a per-device report.

    ``Result.diff`` contains the full dry-run detail from each sub-task, keyed by
    sub-task name (``inventory``, ``interfaces``, ``vrfs``, ``vlans``,
    ``prefixes``, ``ip_addresses``, ``bgp_peerings``, ``bgp_communities``,
    ``vrrp``).

    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_inventory (bool): Check device inventory sync state. Defaults to True.
        check_interfaces (bool): Check interface sync state. Defaults to True.
        check_vrfs (bool): Check VRF and interface assignment sync state.
        check_vlans (bool): Check VLAN and interface assignment sync state.
        check_prefixes (bool): Check prefix 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.
        check_bgp_communities (bool): Check BGP community sync state. Defaults to True.
        check_vrrp (bool): Check VRRP sync state. Defaults to True.
        ignore_deletions (bool): Treat deletion-only differences as in sync.
            Defaults to False.
        sync_kwargs: Per-task arguments keyed by sync task name, or an
            ``nf://`` YAML file containing that dictionary. Supported keys
            are ``sync_device_inventory``, ``sync_device_interfaces``,
            ``sync_vrfs``, ``sync_vlans``, ``sync_device_prefixes``,
            ``sync_device_ip``, ``sync_vrrp``, ``sync_bgp_community``, and
            ``sync_bgp_peerings``.
        **kwargs: Nornir host filter arguments (e.g. ``FL``, ``FC``, ``FB``).

    Returns:
        Result: Per-device summary with every category present. True means
        no drift, False means drift or failure, and None means disabled.
        Enabled checks continue after failures; errors and failed resources
        are propagated. Diff contains each child dry-run diff, or None for
        disabled checks. In_sync considers enabled checks only::

            {
                "<device>": {
                    "in_sync": True | False,
                    "inventory":     True | False | None,
                    "interfaces":    True | False | None,
                    "vrfs":          True | False | None,
                    "vlans":         True | False | None,
                    "prefixes":      True | False | None,
                    "ip_addresses":  True | False | None,
                    "bgp_peerings":  True | False | None,
                    "bgp_communities": True | False | None,
                    "vrrp":          True | False | None,
                }
            }
    """
    devices = devices or []
    instance = instance or self.default_instance
    if isinstance(sync_kwargs, str):
        sync_kwargs = yaml.safe_load(
            self.fetch_file(sync_kwargs, raise_on_fail=True)
        )
    sync_kwargs = sync_kwargs or {}
    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)")
    categories = (
        "inventory",
        "prefixes",
        "interfaces",
        "vrfs",
        "vlans",
        "ip_addresses",
        "vrrp",
        "bgp_communities",
        "bgp_peerings",
    )
    ret.diff = dict.fromkeys(categories)
    # initialize per-device result structure
    for device in devices:
        ret.result[device] = dict.fromkeys(categories)

    check_results = {}

    # --- check inventory ---
    if check_inventory:
        job.event("checking device inventory sync state")
        inventory_result = self.sync_device_inventory(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
            **(sync_kwargs.get("sync_device_inventory") or {}),
        )
        if inventory_result.errors:
            ret.errors.extend(inventory_result.errors)
        for device, data in inventory_result.diff.items():
            ret.result.setdefault(device, {})["inventory"] = (
                not sync_diff_has_changes(data, ignore_deletions=ignore_deletions)
            )
        ret.diff["inventory"] = inventory_result.diff
        check_results["inventory"] = inventory_result

    # --- 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,
            **(sync_kwargs.get("sync_device_interfaces") or {}),
        )
        if intf_result.errors:
            ret.errors.extend(intf_result.errors)
        for device, data in intf_result.diff.items():
            ret.result.setdefault(device, {})["interfaces"] = (
                not sync_diff_has_changes(data, ignore_deletions=ignore_deletions)
            )
        ret.diff["interfaces"] = intf_result.diff
        check_results["interfaces"] = intf_result

    # --- check VRFs and their interface assignments ---
    if check_vrfs:
        job.event("checking VRF sync state")
        vrf_result = self.sync_vrfs(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
            **(sync_kwargs.get("sync_vrfs") or {}),
        )
        if vrf_result.errors:
            ret.errors.extend(vrf_result.errors)
        global_vrf_diff = {
            section: vrf_result.diff.get(section, {})
            for section in ("vrfs", "route_targets", "routing_policies")
        }
        for device in devices:
            device_diff = {
                **global_vrf_diff,
                "interfaces": vrf_result.diff.get("interfaces", {}).get(device, {}),
            }
            ret.result[device]["vrfs"] = not sync_diff_has_changes(
                device_diff, ignore_deletions=ignore_deletions
            )
        ret.diff["vrfs"] = vrf_result.diff
        check_results["vrfs"] = vrf_result

    # --- check VLANs and their interface assignments ---
    if check_vlans:
        job.event("checking VLAN sync state")
        vlan_result = self.sync_vlans(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
            **(sync_kwargs.get("sync_vlans") or {}),
        )
        if vlan_result.errors:
            ret.errors.extend(vlan_result.errors)
        for device in devices:
            device_diff = {
                "vlans": vlan_result.diff.get("vlans", {}),
                "interfaces": vlan_result.diff.get("interfaces", {}).get(
                    device, {}
                ),
            }
            ret.result[device]["vlans"] = not sync_diff_has_changes(
                device_diff, ignore_deletions=ignore_deletions
            )
        ret.diff["vlans"] = vlan_result.diff
        check_results["vlans"] = vlan_result

    # --- check prefixes ---
    if check_prefixes:
        job.event("checking prefix sync state")
        prefix_result = self.sync_device_prefixes(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
            **(sync_kwargs.get("sync_device_prefixes") or {}),
        )
        if prefix_result.errors:
            ret.errors.extend(prefix_result.errors)
        prefixes_in_sync = not sync_diff_has_changes(
            prefix_result.diff, ignore_deletions=ignore_deletions
        )
        for device in devices:
            ret.result[device]["prefixes"] = prefixes_in_sync
        ret.diff["prefixes"] = prefix_result.diff
        check_results["prefixes"] = prefix_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,
            **(sync_kwargs.get("sync_device_ip") or {}),
        )
        if ip_result.errors:
            ret.errors.extend(ip_result.errors)
        for device, data in ip_result.diff.items():
            ret.result.setdefault(device, {})["ip_addresses"] = (
                not sync_diff_has_changes(data, ignore_deletions=ignore_deletions)
            )
        ret.diff["ip_addresses"] = ip_result.diff
        check_results["ip_addresses"] = ip_result

    # --- check VRRP ---
    if check_vrrp:
        job.event("checking VRRP sync state")
        vrrp_result = self.sync_vrrp(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
            **(sync_kwargs.get("sync_vrrp") or {}),
        )
        if vrrp_result.errors:
            ret.errors.extend(vrrp_result.errors)
        for device, data in vrrp_result.diff.items():
            ret.result.setdefault(device, {})["vrrp"] = not sync_diff_has_changes(
                data, ignore_deletions=ignore_deletions
            )
        ret.diff["vrrp"] = vrrp_result.diff
        check_results["vrrp"] = vrrp_result

    # --- check BGP communities ---
    if check_bgp_communities:
        job.event("checking BGP community sync state")
        community_result = self.sync_bgp_community(
            job=job,
            instance=instance,
            dry_run=True,
            timeout=timeout,
            devices=list(devices),
            branch=branch,
            **(sync_kwargs.get("sync_bgp_community") or {}),
        )
        if community_result.errors:
            ret.errors.extend(community_result.errors)
        communities_in_sync = not sync_diff_has_changes(
            community_result.diff, ignore_deletions=ignore_deletions
        )
        for device in devices:
            ret.result[device]["bgp_communities"] = communities_in_sync
        ret.diff["bgp_communities"] = community_result.diff
        check_results["bgp_communities"] = community_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,
            **(sync_kwargs.get("sync_bgp_peerings") or {}),
        )
        if bgp_result.errors:
            ret.errors.extend(bgp_result.errors)
        for device, data in bgp_result.diff.items():
            ret.result.setdefault(device, {})["bgp_peerings"] = (
                not sync_diff_has_changes(data, ignore_deletions=ignore_deletions)
            )
        ret.diff["bgp_peerings"] = bgp_result.diff
        check_results["bgp_peerings"] = bgp_result

    for category, child_result in check_results.items():
        ret.failed = ret.failed or child_result.failed
        ret.resources_failed = sorted(
            set(ret.resources_failed) | set(child_result.resources_failed)
        )
        for device in devices:
            if device in child_result.resources_failed or (
                child_result.failed and not child_result.resources_failed
            ):
                ret.result[device][category] = False
            elif ret.result[device][category] is None:
                ret.result[device][category] = False
                ret.errors.append(f"{device}: no {category} check data returned")

    checked_categories = {
        "inventory": check_inventory,
        "interfaces": check_interfaces,
        "vrfs": check_vrfs,
        "vlans": check_vlans,
        "prefixes": check_prefixes,
        "ip_addresses": check_ip_addresses,
        "vrrp": check_vrrp,
        "bgp_communities": check_bgp_communities,
        "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