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"