Sync VLANs

The sync_vlans task reconciles VLAN object names and descriptions from live devices into NetBox. It requests the normalized TTP vlans getter once for the validated device set, aggregates all successful worker results, then compares VLANs by VID within each VLAN-group or site scope. Names and descriptions are synchronized values and do not form part of a VLAN's identity.

Ordered vlan_map rules place matching VLANs into existing VLAN groups. VLANs which match no rule use the scalar vlan_group when supplied, otherwise they use their device site. VLAN groups are recommended for new deployments because direct VLAN-to-site assignment is deprecated in NetBox 4.4.

This task differs from sync_device_interfaces: sync_vlans manages VLAN names and descriptions, while sync_device_interfaces manages interface objects and VLAN associations and may create placeholder VLANs. Run sync_vlans first when possible. A later sync_vlans run updates placeholders created by interface synchronization.

Inputs

Input	Default	Description
instance	Worker default	NetBox instance name.
dry_run	False	Calculate the diff without writing to NetBox.
with_approval	False	Present the prepared plan for approval before writing. Ignored during dry-run.
timeout	600	Timeout in seconds for host resolution and TTP parsing.
devices	None	Explicit NetBox and Nornir device names.
branch	None	NetBox Branching plugin branch name.
vlan_group	None	Existing group for live VLANs not matched by vlan_map.
vlan_map	None	Ordered rules mapping live VLANs to existing groups.
filter_by_vlan_ids	None	VLAN IDs or inclusive ranges such as 100 and 200-299.
Nornir filters	None	FO, FB, FH, FC, FR, FG, FP, FL, FM, FX, and FN.

Provide devices or at least one Nornir host filter.

VLAN mapping

Each rule contains an exact NetBox VLAN group name. Additional matching criteria are optional:

- vlan_group: CAMPUS
  vlan_ids:
    - 100-199
  vlan_names:
    - USERS*
    - VOICE*
  device_names:
    - leaf-*
  interface_names:
    - Ethernet*

Rules are evaluated in list order and the first match wins. Values inside one criterion use OR logic. Populated criteria use AND logic. VLAN and device names use case-sensitive glob matching. VLAN ranges are inclusive and must remain within 1..4094. VLAN sync ignores interface_names because its live VLAN records have no interface context. Every rule also uses its NetBox VLAN group's configured vid_ranges; explicit vlan_ids narrow those ranges. An unmatched VLAN uses vlan_group when supplied, otherwise it uses its device site.

Every group referenced by vlan_map or vlan_group must already exist. The task resolves groups by exact name and does not create or update groups. Mapping rules are constrained by their group's configured VID ranges. The scalar vlan_group is an unconditional fallback and does not filter live VLANs by the group's configured ranges; NetBox validates the resulting writes.

Live data

The task runs Nornir parse_ttp with get="vlans", which returns:

- vid: 100
  name: USERS
  description: User access VLAN

Only vid, name, and description are managed. Names and descriptions are trimmed, null descriptions become an empty string, and case is preserved. filter_by_vlan_ids removes out-of-range records from both the complete live device dataset and NetBox before comparison.

Identical observations from multiple devices in one scope are collapsed. VLAN observations with the same VID but different names or descriptions are reported as source conflicts. The first device in sorted device-name order supplies the values to synchronize; each later conflicting device is identified in errors. A conflict does not fail or skip that VLAN: the task still creates or updates it using the first device's values. Results returned for the same device by multiple Nornir workers are aggregated before identical observations are collapsed.

Output

Results are keyed by scope, for example group:CAMPUS or site:NORFAB-LAB.

Dry-run returns the standard sync diff shape:

{
  "site:NORFAB-LAB": {
    "create": [110],
    "update": {
      "210": {
        "name": {
          "old_value": "VLAN_210",
          "new_value": "USERS"
        }
      }
    },
    "delete": [],
    "in_sync": [310]
  }
}

Live runs use completed-action verbs. The prepared plan remains available in the top-level diff field:

{
  "site:NORFAB-LAB": {
    "created": [110],
    "updated": [210],
    "deleted": [],
    "in_sync": [310]
  }
}

NetBox VLANs are matched only by VID and group or site scope. An existing VLAN with a stale name is therefore updated directly. The task assumes each scope contains at most one VLAN for a given VID; VLAN groups already enforce this uniqueness.

Deletions

The task does not delete VLANs. Live parsing does not provide a reliable way to identify which additional NetBox VLANs are stale. The standard result shape therefore retains empty delete and deleted lists.

Branches

The task obtains its NetBox client with the requested branch. Reads and writes therefore remain inside that branch. The NetBox Branching plugin must be installed and configured for branch use.

Examples

CLIPython

Preview site-scoped VLAN changes:

nf# netbox sync vlans devices fn-ceos-lf-1 dry-run

Select devices with a Nornir filter and restrict VLAN IDs:

nf# netbox sync vlans FC leaf vlan-ids 100 200-299

Place all unmatched VLANs into one existing group:

nf# netbox sync vlans devices fn-ceos-lf-1 vlan-group CAMPUS

from norfab.core.nfapi import NorFab

with NorFab(inventory="./inventory.yaml") as nf:
    client = nf.make_client()
    result = client.run_job(
        "netbox",
        "sync_vlans",
        workers="any",
        kwargs={
            "devices": ["fn-ceos-lf-1", "fn-ceos-lf-2"],
            "dry_run": True,
            "filter_by_vlan_ids": ["100-399"],
            "vlan_map": [
                {
                    "vlan_group": "CAMPUS",
                    "vlan_ids": ["100-199"],
                    "vlan_names": ["TEST_L*"],
                    "device_names": ["fn-ceos-lf-*"],
                }
            ],
        },
    )
    print(result)

Troubleshooting

Missing parser data

Confirm the device platform is supported by the TTP vlans getter and that the Nornir worker can run the getter's command. Missing or malformed results are reported as errors; valid results from other devices and workers still proceed.

Live VLAN conflicts

Select devices that share one authoritative VLAN definition or correct their name and description differences. The first device in sorted order is used and each later conflicting device is listed in errors. The task continues to synchronize the first device's values.

VLAN group resolution

Check that the scalar vlan_group and every group named in vlan_map exist exactly as written. Group resolution finishes before live collection or writes.

NetBox bulk failures

NetBox validates bulk creates and updates atomically. Correct the reported validation or dependency error and rerun the dry-run. A write failure aborts the task and is not reported as an applied action.

Task command shell reference

nf# man tree netbox.sync.vlans

R - required field, M - supports multiline input, D - dynamic key

root
└── netbox:    Netbox service
    └── sync:    Sync Netbox data
        └── vlans:    Sync live VLAN configuration with NetBox
            ├── instance:    Netbox instance name to target
            ├── dry-run:    Calculate the VLAN diff without writing to NetBox, default 'False'
            ├── branch:    NetBox branching plugin branch name to use
            ├── 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
            ├── FM:    Filter hosts by platform
            ├── FX:    Filter hosts excluding them by name
            ├── FN:    Negate the match
            ├── devices:    List of NetBox devices to collect VLANs from
            ├── timeout:    Job timeout
            ├── with-approval:    Preview VLAN changes and ask for review before writing to NetBox, default 'False'
            ├── vlan-group:    Exact group name for live VLANs not matched by vlan-map
            ├── vlan-map:    Ordered rules mapping live VLANs to NetBox VLAN groups
            ├── vlan-ids:    VLAN IDs or inclusive ranges to reconcile
            ├── workers:    Filter worker to target, default 'any'
            ├── verbose-result:    Control output details, default 'False'
            └── nowait:    Do not wait for job to complete, default 'False'
nf#

Python API reference

Synchronize live VLAN names and descriptions with NetBox.

VLANs are mapped by the first matching vlan_map rule. Rule criteria match VLAN IDs, VLAN names, and device names; populated criteria are combined with AND. Interface name criteria are ignored because VLAN records have no interface context. VLANs which match no rule use vlan_group when supplied, otherwise they use their device site.

Parameters:

Name	Type	Description	Default
job	Job	NorFab job object.	required
instance	Union[None, str]	NetBox instance name. Uses the default instance when omitted.	None
dry_run	bool	Return the calculated diff without writing to NetBox.	False
with_approval	bool	Ask for approval before applying the prepared diff.	False
timeout	int	Timeout in seconds for Nornir host resolution and parsing.	600
devices	Union[None, list]	Explicit NetBox and Nornir device names.	None
branch	Union[None, str]	NetBox Branching plugin branch name.	None
vlan_group	Union[None, str]	Group for VLANs not matched by vlan_map.	None
vlan_map	Union[None, list]	Ordered VLAN-to-group mapping rules.	None
filter_by_vlan_ids	Union[None, list[str]]	VLAN IDs or inclusive ranges to reconcile.	None
**kwargs	Any	Nornir FFun host filters.	{}

Returns:

Name	Type	Description
Result	Result	Scope-keyed VLAN synchronization actions.

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

@Task(
    fastapi={"methods": ["POST"], "schema": NetboxFastApiArgs.model_json_schema()},
    input=SyncVlansInput,
    output=SyncVlansResult,
    mcp={
        "annotations": {
            "title": "Sync VLANs",
            "readOnlyHint": False,
            "destructiveHint": True,
            "idempotentHint": True,
            "openWorldHint": True,
        }
    },
)
def sync_vlans(
    self,
    job: Job,
    instance: Union[None, str] = None,
    dry_run: bool = False,
    with_approval: bool = False,
    timeout: int = 600,
    devices: Union[None, list] = None,
    branch: Union[None, str] = None,
    vlan_group: Union[None, str] = None,
    vlan_map: Union[None, list] = None,
    filter_by_vlan_ids: Union[None, list[str]] = None,
    **kwargs: Any,
) -> Result:
    """Synchronize live VLAN names and descriptions with NetBox.

    VLANs are mapped by the first matching ``vlan_map`` rule. Rule criteria
    match VLAN IDs, VLAN names, and device names; populated criteria are
    combined with AND. Interface name criteria are ignored because VLAN
    records have no interface context. VLANs which match no rule use
    ``vlan_group`` when supplied, otherwise they use their device site.

    Args:
        job: NorFab job object.
        instance: NetBox instance name. Uses the default instance when omitted.
        dry_run: Return the calculated diff without writing to NetBox.
        with_approval: Ask for approval before applying the prepared diff.
        timeout: Timeout in seconds for Nornir host resolution and parsing.
        devices: Explicit NetBox and Nornir device names.
        branch: NetBox Branching plugin branch name.
        vlan_group: Group for VLANs not matched by ``vlan_map``.
        vlan_map: Ordered VLAN-to-group mapping rules.
        filter_by_vlan_ids: VLAN IDs or inclusive ranges to reconcile.
        **kwargs: Nornir FFun host filters.

    Returns:
        Result: Scope-keyed VLAN synchronization actions.
    """
    devices = list(devices or [])
    instance = instance or self.default_instance
    ret = Result(
        task=f"{self.name}:sync_vlans",
        result={},
        resources=[instance],
        dry_run=dry_run,
        diff={},
    )

    job.event(
        f"starting VLAN sync using NetBox instance '{instance}' for "
        f"{len(devices)} explicit device(s)"
    )
    log.info(f"{self.name} - Sync VLANs: instance '{instance}', dry_run={dry_run}")
    nb = self._get_pynetbox(instance, branch=branch, job=job)

    # Resolve and validate the complete device set.
    if kwargs:
        job.event("resolving devices from Nornir filters")
        devices.extend(self.get_nornir_hosts(kwargs, timeout))
    devices = sorted(set(devices))
    if not devices:
        msg = "no devices specified"
        job.event(msg, severity="ERROR")
        log.error(f"{self.name} - Sync VLANs: {msg}")
        ret.errors.append(msg)
        ret.failed = True
        return ret
    job.event(f"selected {len(devices)} device(s)")

    nb_devices = {
        device.name: device
        for device in self.bulk_filter(
            nb.dcim.devices,
            name=devices,
            fields="id,name,site",
        )
    }
    missing_devices = [name for name in devices if name not in nb_devices]
    for device_name in missing_devices:
        msg = f"device '{device_name}' not found in NetBox"
        job.event(msg, severity="ERROR")
        log.error(f"{self.name} - Sync VLANs: {msg}")
        ret.errors.append(msg)
    devices = [name for name in devices if name in nb_devices]
    if not devices:
        ret.failed = True
        return ret
    job.event(f"validated {len(devices)} NetBox device(s)")

    # Expand filters and resolve every VLAN group before collecting live data.
    expanded_filter = {
        int(vlan_id)
        for vlan_range in filter_by_vlan_ids or []
        for vlan_id in expand_alphanumeric_range(f"[{vlan_range}]")
    }
    rules = prepare_vlan_map(vlan_map)
    job.event(
        f"prepared {len(rules)} VLAN map rule(s) and "
        f"{len(expanded_filter)} VLAN filter ID(s)"
    )

    vlan_groups = {}
    group_names = [rule["vlan_group"] for rule in rules]
    if vlan_group:
        group_names.append(vlan_group)
    for group_name in dict.fromkeys(group_names):
        group = nb.ipam.vlan_groups.get(name=group_name)
        if group is None:
            msg = f"vlan group '{group_name}' does not exist in NetBox"
            job.event(msg, severity="ERROR")
            log.error(f"{self.name} - Sync VLANs: {msg}")
            ret.errors.append(msg)
            ret.failed = True
            return ret
        vlan_groups[group_name] = group
    rules = prepare_vlan_map(rules, vlan_groups)
    if vlan_groups:
        job.event(f"resolved {len(vlan_groups)} NetBox VLAN group(s)")

    # Build the site and group scopes used for comparison.
    scope_metadata = {}
    site_scopes = {}
    group_scopes = {}
    for device_name in devices:
        device = nb_devices[device_name]
        scope = f"site:{device.site.name}"
        site_scopes[device_name] = scope
        scope_metadata[scope] = {
            "type": "site",
            "id": device.site.id,
            "name": device.site.name,
        }
    for group in vlan_groups.values():
        scope = f"group:{group.name}"
        group_scopes[group.name] = scope
        scope_metadata[scope] = {
            "type": "group",
            "id": group.id,
            "name": group.name,
        }
    job.event(f"resolved {len(scope_metadata)} VLAN scope(s)")

    # Collect VLANs once from all Nornir workers and normalize each response.
    job.event(f"collecting live VLANs from {len(devices)} device(s)")
    log.info(f"{self.name} - Sync VLANs: collecting from {len(devices)} device(s)")
    parse_data = self.client.run_job(
        "nornir",
        "parse_ttp",
        kwargs={"get": "vlans", "FL": devices},
        workers="all",
        timeout=timeout,
    )
    job.event(f"received VLAN data from {len(parse_data)} Nornir worker(s)")
    live_by_device = {}
    for worker_name, worker_data in parse_data.items():
        if worker_data.get("failed"):
            msg = f"worker '{worker_name}' failed to collect live VLAN data"
            job.event(msg, severity="ERROR")
            log.error(f"{self.name} - Sync VLANs: {msg}")
            ret.errors.append(msg)
            continue
        worker_result = worker_data.get("result")
        if not isinstance(worker_result, dict):
            msg = f"worker '{worker_name}' returned a malformed Nornir VLAN result"
            job.event(msg, severity="ERROR")
            log.error(f"{self.name} - Sync VLANs: {msg}")
            ret.errors.append(msg)
            continue
        for device_name, records in worker_result.items():
            if device_name not in nb_devices:
                continue
            device_vlans = live_by_device.setdefault(device_name, [])
            if not isinstance(records, list):
                msg = f"device '{device_name}' VLAN parsing result is not a list"
                job.event(msg, severity="ERROR")
                log.error(f"{self.name} - Sync VLANs: {msg}")
                ret.errors.append(msg)
                continue
            for record in records:
                if expanded_filter and record["vid"] not in expanded_filter:
                    continue
                device_vlans.append(
                    {
                        "vid": record["vid"],
                        "name": record["name"].strip(),
                        "description": (record.get("description") or "").strip(),
                    }
                )

    for device_name in devices:
        if device_name not in live_by_device:
            msg = f"device '{device_name}' is missing a live VLAN result"
            job.event(msg, severity="ERROR")
            log.error(f"{self.name} - Sync VLANs: {msg}")
            ret.errors.append(msg)
    if not live_by_device:
        log.error(f"{self.name} - Sync VLANs: no usable live VLAN data")
        ret.failed = True
        return ret
    parsed_count = sum(len(records) for records in live_by_device.values())
    job.event(
        f"parsed {parsed_count} live VLAN record(s) from "
        f"{len(live_by_device)} device(s)"
    )

    # Map live observations to a group or site. A VID uniquely identifies a
    # VLAN within either scope; name and description are synchronized fields.
    observations = {scope: {} for scope in scope_metadata}
    for device_name in sorted(live_by_device):
        for vlan in live_by_device[device_name]:
            selected_group_name = (
                match_vlan_map(
                    rules,
                    vlan_id=vlan["vid"],
                    vlan_name=vlan["name"],
                    device_name=device_name,
                    interface_name=None,
                )
                or vlan_group
            )
            if selected_group_name:
                scope = group_scopes[selected_group_name]
            else:
                scope = site_scopes[device_name]
            observations[scope].setdefault(vlan["vid"], []).append(
                {"device": device_name, **vlan}
            )

    # Collapse identical observations. The first device in sorted order is
    # authoritative; conflicting values from later devices are reported.
    normalized_live = {scope: {} for scope in scope_metadata}
    source_conflict_count = 0
    for scope in sorted(observations):
        for vid in sorted(observations[scope]):
            records = observations[scope][vid]
            desired = records[0]
            desired_values = (desired["name"], desired["description"])
            for conflicting in records[1:]:
                if (
                    conflicting["name"],
                    conflicting["description"],
                ) == desired_values:
                    continue
                msg = (
                    f"{scope} VLAN {vid} source conflict: using VLAN name "
                    f"'{desired['name']}' from device '{desired['device']}'; "
                    f"conflicting device '{conflicting['device']}' reports "
                    f"VLAN name '{conflicting['name']}'"
                )
                job.event(msg, severity="ERROR")
                log.error(f"{self.name} - Sync VLANs: {msg}")
                ret.errors.append(msg)
                source_conflict_count += 1
            normalized_live[scope][vid] = {
                "vid": desired["vid"],
                "name": desired["name"],
                "description": desired["description"],
            }

    # Load current NetBox VLANs once, keyed only by VID within each scope.
    job.event(f"loading NetBox VLANs from {len(scope_metadata)} scope(s)")
    normalized_netbox = {scope: {} for scope in scope_metadata}
    netbox_objects = {scope: {} for scope in scope_metadata}
    netbox_vlan_count = 0
    for scope in sorted(scope_metadata):
        metadata = scope_metadata[scope]
        scope_filter = {f"{metadata['type']}_id": metadata["id"]}
        for vlan in self.bulk_filter(
            nb.ipam.vlans,
            fields="id,vid,name,description",
            **scope_filter,
        ):
            vid = int(vlan.vid)
            if expanded_filter and vid not in expanded_filter:
                continue
            netbox_vlan_count += 1
            normalized_netbox[scope][vid] = {
                "vid": vid,
                "name": str(vlan.name).strip(),
                "description": str(
                    getattr(vlan, "description", None) or ""
                ).strip(),
            }
            netbox_objects[scope][vid] = vlan
    job.event(f"loaded {netbox_vlan_count} in-scope NetBox VLAN record(s)")

    # Compare the normalized VID-keyed structures.
    job.event("calculating VLAN sync diff")
    internal_diff = self.make_diff(normalized_live, normalized_netbox)

    full_diff = {
        scope: {
            "create": sorted(actions["create"]),
            "update": {
                str(vid): actions["update"][vid]
                for vid in sorted(actions["update"])
            },
            "delete": [],
            "in_sync": sorted(actions["in_sync"]),
        }
        for scope, actions in sorted(internal_diff.items())
    }
    create_count = sum(len(actions["create"]) for actions in full_diff.values())
    update_count = sum(len(actions["update"]) for actions in full_diff.values())
    in_sync_count = sum(len(actions["in_sync"]) for actions in full_diff.values())
    job.event(
        "vlan sync diff complete: "
        f"{create_count} create, {update_count} update, "
        f"{in_sync_count} in sync, "
        f"{source_conflict_count} source conflict(s)"
    )

    if dry_run:
        job.event("dry-run requested, returning VLAN sync diff without changes")
        log.info(f"{self.name} - Sync VLANs: dry-run complete")
        ret.result = full_diff
        ret.dry_run = True
        return ret
    if with_approval:
        job.event("requesting approval for the prepared VLAN sync plan")
    if with_approval and not review_sync_task_result(job, "VLAN sync", full_diff):
        ret.status = "skipped"
        ret.result = full_diff
        ret.dry_run = True
        ret.messages.append("review declined; changes were not applied")
        log.info(f"{self.name} - Sync VLANs: approval declined")
        return ret

    # Apply each scope independently using NetBox bulk operations.
    ret.diff = full_diff
    ret.result = {
        scope: {
            "created": [],
            "updated": [],
            "deleted": [],
            "in_sync": actions["in_sync"],
        }
        for scope, actions in full_diff.items()
    }
    for scope in sorted(internal_diff):
        actions = internal_diff[scope]
        metadata = scope_metadata[scope]
        create_vids = sorted(actions["create"])
        update_vids = sorted(actions["update"])
        job.event(
            f"applying {scope}: {len(create_vids)} create, "
            f"{len(update_vids)} update"
        )

        scope_arg = {f"{metadata['type']}_id": metadata["id"]}
        create_payloads = [
            build_vlan_payload(
                **normalized_live[scope][vid],
                **scope_arg,
            )
            for vid in create_vids
        ]
        if create_payloads:
            nb.ipam.vlans.create(create_payloads)
            ret.result[scope]["created"].extend(create_vids)
            job.event(f"{scope}: created {len(create_payloads)} VLAN(s)")

        update_payloads = []
        for vid in update_vids:
            field_changes = actions["update"][vid]
            desired = normalized_live[scope][vid]
            payload = {"id": netbox_objects[scope][vid].id}
            for field in ("name", "description"):
                if field in field_changes:
                    payload[field] = desired[field]
            update_payloads.append(payload)
        if update_payloads:
            nb.ipam.vlans.update(update_payloads)
            ret.result[scope]["updated"].extend(update_vids)
            job.event(f"{scope}: updated {len(update_payloads)} VLAN(s)")
        job.event(f"completed VLAN changes for {scope}")

    job.event("vlan sync complete")
    log.info(
        f"{self.name} - Sync VLANs complete: {create_count} created, "
        f"{update_count} updated, {in_sync_count} in sync"
    )
    return ret