vmware: derive the NetBox VM name and description with vm_name_regex

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,

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<name>[^ (]+)(?: \\((?P<description>.*)\\))?$"),
             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()

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:

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<name>[^ (]+)(?: \((?P<description>.*)\))?$
+
 ; tag source options
 
 ; sync tags assigned to clusters, hosts and VMs in vCenter to NetBox

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: <https://opensource.org/licenses/MIT>.
+
+"""
+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<name>[^ (]+)(?: \((?P<description>.*)\))?$"
+
+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>[^ (]+)(?: \(.*\))?$"))
+
+    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>.+)$"))
+
+    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"^(?P<name>vm[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<name>[^ (]*)(?: \((?P<description>.*)\))?$"))
+
+    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"(?P<name>vm\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<name>\S+) (?P<description>.*)$"))
+
+    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<name>"\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 = ^(?P<name>vm\\d+)_(?P<description>.*)$\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"