From b17d3f5d8e3af01811522ba467da9bc1a2599f07 Mon Sep 17 00:00:00 2001 From: Sergey Sannikov Date: Sat, 10 Oct 2026 04:42:34 +0400 Subject: [PATCH] vmware: derive the NetBox VM name and description with vm_name_regex The new option vm_name_regex is applied to the vCenter VM name before any other name handling. Its named group "name" becomes the NetBox VM name and an optional named group "description" fills the VM's description. A name the regex does not match is used unchanged, and so is a name whose "name" group takes no part in the match or matches only whitespace; strip_vm_domain_name still runs afterwards. The regex is compiled when the config is loaded and must contain the "name" group. The resulting name is what the VM filters and relations see and what NetBox VMs are matched by, so it has to stay unique per cluster; the warning about a duplicate VM name says so when the option is in use. A name or description longer than the NetBox field is reported before it is cut. The description field is added to the VM data model and is only sent to NetBox 3.4 and later, which is where the field appeared. Fixes #465 --- module/netbox/object_classes.py | 1 + module/sources/vmware/config.py | 30 ++++ module/sources/vmware/connection.py | 75 +++++++++- settings-example.ini | 11 ++ tests/test_vmware_vm_name_regex.py | 211 ++++++++++++++++++++++++++++ 5 files changed, 324 insertions(+), 4 deletions(-) create mode 100644 tests/test_vmware_vm_name_regex.py diff --git a/module/netbox/object_classes.py b/module/netbox/object_classes.py index e9fd3fb..a26cf46 100644 --- a/module/netbox/object_classes.py +++ b/module/netbox/object_classes.py @@ -2000,6 +2000,7 @@ def __init__(self, *args, **kwargs): "memory": int, "disk": int, "comments": str, + "description": 200, "primary_ip4": NBIPAddress, "primary_ip6": NBIPAddress, "site": NBSite, diff --git a/module/sources/vmware/config.py b/module/sources/vmware/config.py index c111a54..771eef4 100644 --- a/module/sources/vmware/config.py +++ b/module/sources/vmware/config.py @@ -417,6 +417,19 @@ def __init__(self): bool, description="strip domain part from VM name before syncing VM to NetBox", default_value=False), + ConfigOption("vm_name_regex", + str, + description="""a regex expression applied to the vCenter VM name before any other + name handling ('strip_vm_domain_name' runs after it). It must contain a named + group 'name'; the match of that group becomes the NetBox VM name. An optional + named group 'description' sets the VM description field (NetBox 3.4 and later) + when it matches something; a description already set in NetBox is kept + otherwise, like comments are. If the regex does not match, or the 'name' group + matches nothing, the name is used unchanged. The resulting name is the one all + other VM options see ('vm_include_filter', 'vm_exclude_filter', the + 'vm_*_relation' options, 'vm_exclude_disk_sync') and the one NetBox VMs are + matched by, so it has to stay unique per cluster.""", + config_example="^(?P[^ (]+)(?: \\((?P.*)\\))?$"), ConfigOptionGroup(title="tag source", description="""\ sync tags assigned to clusters, hosts and VMs in vCenter to NetBox @@ -720,6 +733,23 @@ def validate_options(self): continue + if option.key == "vm_name_regex": + + re_compiled = None + try: + re_compiled = re.compile(option.value) + except Exception as e: + log.error(f"Problem parsing regular expression for '{self.source_name}.{option.key}': {e}") + self.set_validation_failed() + + if re_compiled is not None and "name" not in re_compiled.groupindex: + log.error(f"Config option '{option.key}' must contain a named group 'name'.") + self.set_validation_failed() + + option.set_value(re_compiled) + + continue + if "relation" in option.key and "vlan_group_relation" not in option.key: relation_data = list() diff --git a/module/sources/vmware/connection.py b/module/sources/vmware/connection.py index 6590964..6c5d255 100644 --- a/module/sources/vmware/connection.py +++ b/module/sources/vmware/connection.py @@ -103,6 +103,11 @@ class VMWareHandler(SourceBase): # vCenter itself allows far fewer object_path_max_depth = 20 + # field lengths of the NetBox VM model (see NBVM.data_model), used to warn about + # names and descriptions vm_name_regex derives that NetBox would cut + vm_name_max_length = 64 + vm_description_max_length = 200 + recursion_level = 0 # internal vars @@ -3194,6 +3199,61 @@ def add_host(self, obj): return + def get_vm_name_and_description(self, name): + """ + Apply the configured VM name handling to a vCenter VM name and return the + resulting NetBox name and an optional description. + + 'vm_name_regex' is applied before any other name handling. Its named group + 'name' becomes the NetBox VM name and an optional named group 'description' + sets the VM description. If the regex does not match, or its 'name' group + takes no part in the match or matches only whitespace, the name is used + unchanged. 'strip_vm_domain_name' runs afterwards. + + Parameters + ---------- + name: str + VM name as reported by vCenter + + Returns + ------- + tuple: (name, description) + """ + + vm_description = None + + # apply vm_name_regex before any other name handling + if name is not None and self.settings.vm_name_regex is not None: + regex_match = self.settings.vm_name_regex.match(name) + if regex_match is None: + log.debug2(f"VM name '{name}' does not match vm_name_regex. Using the name unchanged.") + else: + regex_name = regex_match.group("name") + vm_description = regex_match.groupdict().get("description") + + # an optional group takes no part in the match and returns None + if regex_name is None or regex_name.strip() == "": + log.warning(f"The 'name' group of vm_name_regex matched nothing in VM name '{name}'. " + f"Using the name unchanged.") + else: + name = regex_name + + if vm_description is not None and vm_description.strip() == "": + vm_description = None + + if len(name) > self.vm_name_max_length: + log.warning(f"VM name '{name}' derived by vm_name_regex is longer than " + f"{self.vm_name_max_length} characters and will be cut") + + if vm_description is not None and len(vm_description) > self.vm_description_max_length: + log.warning(f"VM description '{vm_description}' derived by vm_name_regex is longer than " + f"{self.vm_description_max_length} characters and will be cut") + + if name is not None and self.settings.strip_vm_domain_name is True: + name = name.split(".")[0] + + return name, vm_description + def add_virtual_machine(self, obj): """ Parse a vCenter VM add to NetBox once all data is gathered. @@ -3223,8 +3283,7 @@ def add_virtual_machine(self, obj): name = get_string_or_none(grab(obj, "name")) - if name is not None and self.settings.strip_vm_domain_name is True: - name = name.split(".")[0] + name, vm_description = self.get_vm_name_and_description(name) # # Filtering @@ -3289,8 +3348,12 @@ def add_virtual_machine(self, obj): cluster_full_name = f"{group.name}/{cluster_name}" if name in self.processed_vm_names.get(cluster_full_name, list()) and obj not in self.objects_to_reevaluate: - log.warning(f"Virtual machine '{name}' for cluster '{cluster_full_name}' already parsed. " - "Make sure to use unique VM names. Skipping") + if self.settings.vm_name_regex is not None: + log.warning(f"Virtual machine '{name}' for cluster '{cluster_full_name}' already parsed. " + "'vm_name_regex' must leave the VM names unique per cluster. Skipping") + else: + log.warning(f"Virtual machine '{name}' for cluster '{cluster_full_name}' already parsed. " + "Make sure to use unique VM names. Skipping") return # add vm to processed list @@ -3415,6 +3478,10 @@ def add_virtual_machine(self, obj): vm_data["platform"] = {"name": platform} if annotation is not None: vm_data["comments"] = annotation + # NetBox got the VM description field with 3.4.0 + if vm_description is not None and \ + version.parse(self.inventory.netbox_api_version) >= version.parse("3.4.0"): + vm_data["description"] = vm_description if tenant_name is not None: vm_data["tenant"] = {"name": tenant_name} if len(vm_tags) > 0: diff --git a/settings-example.ini b/settings-example.ini index cc7af64..d8640c2 100644 --- a/settings-example.ini +++ b/settings-example.ini @@ -416,6 +416,17 @@ password = super-secret ; strip domain part from VM name before syncing VM to NetBox ;strip_vm_domain_name = False +; a regex expression applied to the vCenter VM name before any other name handling +; ('strip_vm_domain_name' runs after it). It must contain a named group 'name'; the match +; of that group becomes the NetBox VM name. An optional named group 'description' sets the +; VM description field (NetBox 3.4 and later) when it matches something; a description +; already set in NetBox is kept otherwise, like comments are. If the regex does not match, +; or the 'name' group matches nothing, the name is used unchanged. The resulting name is +; the one all other VM options see ('vm_include_filter', 'vm_exclude_filter', the +; 'vm_*_relation' options, 'vm_exclude_disk_sync') and the one NetBox VMs are matched by, +; so it has to stay unique per cluster. +;vm_name_regex = ^(?P[^ (]+)(?: \((?P.*)\))?$ + ; tag source options ; sync tags assigned to clusters, hosts and VMs in vCenter to NetBox diff --git a/tests/test_vmware_vm_name_regex.py b/tests/test_vmware_vm_name_regex.py new file mode 100644 index 0000000..1b7dd5d --- /dev/null +++ b/tests/test_vmware_vm_name_regex.py @@ -0,0 +1,211 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2020 - 2026 netbox-sync team. All rights reserved. +# +# netbox-sync.py +# +# This work is licensed under the terms of the MIT license. +# For a copy, see file LICENSE.txt included in this +# repository or visit: . + +""" +Regression tests for issue #465 (vm_name_regex). + +The regex is applied to the vCenter VM name before any other name handling. Its named +group 'name' becomes the NetBox VM name and an optional named group 'description' sets +the VM description. A regex without a 'name' group is rejected at config load time. +""" +import logging +import re +from types import SimpleNamespace + +import pytest + +from module.common.logging import DEBUG2 +from module.netbox.object_classes import NBVM +from module.sources import instantiate_sources +from module.sources.vmware.config import VMWareConfig +from module.sources.vmware.connection import VMWareHandler + +EXAMPLE_REGEX = r"^(?P[^ (]+)(?: \((?P.*)\))?$" + +NETBOX = """ +netbox: + host_fqdn: netbox.example.com + api_token: xyz +""" +SOURCE = """ +source: + vc: + type: vmware + host_fqdn: vcenter.example.com + username: u + password: p +""" + + +def _make_source(vm_name_regex=None, strip_vm_domain_name=False): + # build the handler without its network-bound __init__ + src = object.__new__(VMWareHandler) + src.settings = SimpleNamespace( + vm_name_regex=vm_name_regex, + strip_vm_domain_name=strip_vm_domain_name, + ) + return src + + +def _parse_vmware(load_config, option_lines): + load_config(NETBOX + SOURCE + option_lines, filename="settings.yaml") + handler = VMWareConfig() + handler.source_name = "vc" + return handler.parse(do_log=False) + + +def test_regex_rewrites_name_and_sets_description(): + src = _make_source(vm_name_regex=re.compile(EXAMPLE_REGEX)) + + name, description = src.get_vm_name_and_description("web01 (production)") + + assert name == "web01" + assert description == "production" + + +def test_regex_without_description_group_keeps_description_unset(): + src = _make_source(vm_name_regex=re.compile(r"^(?P[^ (]+)(?: \(.*\))?$")) + + name, description = src.get_vm_name_and_description("web01 (production)") + + assert name == "web01" + assert description is None + + +def test_non_matching_name_is_used_unchanged(caplog): + caplog.set_level(DEBUG2, logger="NetBox-Sync") + src = _make_source(vm_name_regex=re.compile(r"^prefix-(?P.+)$")) + + name, description = src.get_vm_name_and_description("web01") + + assert name == "web01" + assert description is None + assert any("does not match vm_name_regex" in record.getMessage() for record in caplog.records) + + +def test_strip_vm_domain_name_runs_after_the_regex(): + # the regex captures the full name (with domain), strip_vm_domain_name then trims it + src = _make_source(vm_name_regex=re.compile(EXAMPLE_REGEX), strip_vm_domain_name=True) + + name, description = src.get_vm_name_and_description("web01.example.com (production)") + + assert name == "web01" + assert description == "production" + + +def test_name_group_that_takes_no_part_in_the_match_keeps_the_name(caplog): + # an optional name group matches the whole name but returns None for it + src = _make_source(vm_name_regex=re.compile(r"^(?Pvm[0-9]+)?_?.*$")) + + with caplog.at_level(logging.WARNING, logger="NetBox-Sync"): + name, description = src.get_vm_name_and_description("web01") + + assert name == "web01" + assert description is None + assert any("matched nothing" in record.getMessage() for record in caplog.records) + + +def test_blank_name_group_keeps_the_name(): + # the documented example with '*' instead of '+' matches an empty name for "(prod)" + src = _make_source(vm_name_regex=re.compile(r"^(?P[^ (]*)(?: \((?P.*)\))?$")) + + assert src.get_vm_name_and_description(" (prod)") == (" (prod)", "prod") + assert src.get_vm_name_and_description(" ") == (" ", None) + + +def test_empty_description_group_sets_no_description(): + src = _make_source(vm_name_regex=re.compile(EXAMPLE_REGEX)) + + assert src.get_vm_name_and_description("web01 ()") == ("web01", None) + assert src.get_vm_name_and_description("web01 ( )") == ("web01", None) + + +def test_regex_is_anchored_at_the_start_of_the_name(): + # match(), not search(): a pattern without '^' still has to match from the first character + src = _make_source(vm_name_regex=re.compile(r"(?Pvm\d+)")) + + assert src.get_vm_name_and_description("x-vm002") == ("x-vm002", None) + assert src.get_vm_name_and_description("vm002-x") == ("vm002", None) + + +def test_overlong_results_are_reported(caplog): + src = _make_source(vm_name_regex=re.compile(r"^(?P\S+) (?P.*)$")) + + with caplog.at_level(logging.WARNING, logger="NetBox-Sync"): + name, description = src.get_vm_name_and_description("n" * 70 + " " + "d" * 210) + + assert name == "n" * 70 and description == "d" * 210 + messages = [record.getMessage() for record in caplog.records] + assert any("longer than 64" in message for message in messages), messages + assert any("longer than 200" in message for message in messages), messages + + +def test_no_regex_leaves_the_name_untouched(): + src = _make_source() + + name, description = src.get_vm_name_and_description("web01.example.com") + + assert name == "web01.example.com" + assert description is None + + +def test_vm_name_regex_is_unset_by_default(load_config): + settings = _parse_vmware(load_config, "") + + # ConfigOptions answers None for any unknown attribute, so make sure the option exists + assert "vm_name_regex" in vars(settings) + assert settings.vm_name_regex is None + + +def test_regex_with_name_group_is_compiled(load_config): + settings = _parse_vmware(load_config, f" vm_name_regex: '{EXAMPLE_REGEX}'\n") + + assert settings.vm_name_regex is not None + assert "name" in settings.vm_name_regex.groupindex + + +def test_invalid_regex_is_rejected(load_config, caplog): + caplog.set_level(logging.ERROR, logger="NetBox-Sync") + + with pytest.raises(SystemExit): + _parse_vmware(load_config, ' vm_name_regex: "(?P"\n') + + errors = [record.getMessage() for record in caplog.records if record.levelno == logging.ERROR] + assert any("regular expression" in message for message in errors), errors + + +def test_regex_without_name_group_is_rejected(load_config, caplog): + caplog.set_level(logging.ERROR, logger="NetBox-Sync") + + with pytest.raises(SystemExit): + _parse_vmware(load_config, ' vm_name_regex: "^foo.*$"\n') + + errors = [record.getMessage() for record in caplog.records if record.levelno == logging.ERROR] + assert any("named group 'name'" in message for message in errors), errors + + +# +# end to end through vcsim +# + +def test_description_reaches_the_netbox_vm(vcsim, inventory, load_config, vmware_settings): + """The description group lands on the NBVM object, so it is sent to NetBox.""" + if vcsim.name != "vc001": + pytest.skip("asserts VM names of the vc001 inventory") + + load_config(vmware_settings + "\nvm_name_regex = ^(?Pvm\\d+)_(?P.*)$\n") + sources = instantiate_sources() + assert sources and sources[0].init_successful + + inventory.resolve_relations() + sources[0].apply() + + vms = {vm.data.get("name"): vm for vm in inventory.get_all_items(NBVM)} + assert "vm002" in vms and "vm002_xyz-veeam" not in vms + assert vms["vm002"].data.get("description") == "xyz-veeam"