Add VMmanager 6 integration with NetBox

README.md

@@ -290,6 +290,7 @@ Program will exit if all retries failed!
 ## Supported sources
 Check out the documentations for the different sources
 * [vmware](https://github.com/bb-Ricardo/netbox-sync/blob/main/docs/source_vmware.md)
+* [vmmanager](docs/source_vmmanager.md)
 * [check_redfish](https://github.com/bb-Ricardo/netbox-sync/blob/main/docs/source_check_redfish.md)
 
 ### Filtering

module/netbox/connection.py

@@ -580,6 +580,11 @@ def initialize_basic_data(self):
 
         log.debug("Checking/Adding NetBox Sync dependencies")
 
+        # The primary and orphaned tags may already exist in NetBox from a previous
+        # run. Load existing tags first so add_update_object updates/reuses them
+        # instead of trying to POST duplicates (which causes 400 conflicts).
+        self.query_current_data([NBTag])
+
         prune_text = f"Pruning is enabled and Objects will be automatically " \
                      f"removed after {self.settings.prune_delay_in_days} days"
 

module/netbox/object_classes.py

@@ -835,12 +835,24 @@ def get_display_name(self, data=None, including_second_key=False):
                 read_from_netbox = True if secondary_key_value.get("id", 0) != 0 else False
                 secondary_key_value = self.get_display_name(data=secondary_key_value)
 
-            if secondary_key_value is None and read_from_netbox is False and include_secondary_key_if_present is False:
+            # NetBox permits global clusters without a scope. For these objects,
+            # scope_id=None is valid and should not trigger the generic missing-key
+            # warning or become part of the display/matching name. A scoped cluster
+            # with a scope_type but no scope_id remains suspicious and is warned about.
+            is_unscoped_cluster = (
+                self.name == "cluster"
+                and secondary_key == "scope_id"
+                and secondary_key_value is None
+                and this_data_set.get("scope_type") is None
+            )
+
+            if secondary_key_value is None and read_from_netbox is False and include_secondary_key_if_present is False and not is_unscoped_cluster:
                 log.warning(f"Unable to determine second key '{secondary_key}' for {self.name} '{my_name}', "
                             f"got: {org_secondary_key_value}")
                 log.warning("This could cause serious errors and lead to wrongly assigned object relations!!!")
 
-            my_name = f"{my_name} ({secondary_key_value})"
+            if not is_unscoped_cluster:
+                my_name = f"{my_name} ({secondary_key_value})"
 
         return my_name
 
@@ -1950,6 +1962,7 @@ def __init__(self, *args, **kwargs):
             "platform": NBPlatform,
             "serial": 50,
             "site": NBSite,
+            "description": 200,
             "status": ["offline", "active", "planned", "staged", "failed", "inventory", "decommissioning"],
             "cluster": NBCluster,
             "asset_tag": 50,

module/sources/__init__.py

@@ -10,6 +10,7 @@
 # define all available sources here
 from module.sources.vmware.connection import VMWareHandler
 from module.sources.hetzner.connection import HetznerHandler
+from module.sources.vmmanager.connection import VMmanagerHandler
 from module.sources.check_redfish.import_inventory import CheckRedfish
 
 from module.common.logging import get_logger
@@ -19,7 +20,7 @@
 from module.config import source_config_section_name
 
 # list of valid sources
-valid_sources = [VMWareHandler, CheckRedfish, HetznerHandler]
+valid_sources = [VMWareHandler, CheckRedfish, HetznerHandler, VMmanagerHandler]
 
 
 def validate_source(source_class_object=None, state="pre"):

module/sources/common/source_base.py

@@ -341,8 +341,14 @@ def add_update_interface(self, interface_object, device_object, interface_data,
         interface_mac_address = None
         if version.parse(self.inventory.netbox_api_version) >= version.parse("4.2.0") and \
                 interface_data.get("mac_address") is not None:
-            interface_mac_address = interface_data.get("mac_address")
+            raw_mac_address = interface_data.get("mac_address")
+            # Some inventory sources return an empty string for an unknown MAC.
+            # Treat whitespace-only values as missing and never create an invalid
+            # dcim/mac-addresses object from them.
+            interface_mac_address = str(raw_mac_address).strip() if raw_mac_address is not None else None
             del(interface_data["mac_address"])
+            if not interface_mac_address:
+                interface_mac_address = None
 
         ip_tenant_inheritance_order = self.settings.ip_tenant_inheritance_order
 

module/sources/vmmanager/__init__.py

@@ -0,0 +1,2 @@
+# VMmanager inventory source.
+from module.sources.vmmanager.connection import VMmanagerHandler

module/sources/vmmanager/client.py

@@ -0,0 +1,137 @@
+# -*- coding: utf-8 -*-
+# Licensed under the MIT license.
+import time
+from urllib.parse import urljoin
+import requests
+
+class VMmanagerAPIError(RuntimeError):
+    """VMmanager cannot provide a trustworthy inventory snapshot."""
+
+class VMmanagerClient:
+    """Read-only client for the documented VMmanager 6 API."""
+    def __init__(self, base_url, email, password, verify_ssl=True, timeout=30, session=None):
+        self.base_url = base_url.rstrip("/") + "/"
+        self.email, self.password = email, password
+        self.verify_ssl, self.timeout = verify_ssl, max(1, int(timeout))
+        self.session = session or requests.Session()
+        self._token = None
+
+    def _url(self, path):
+        return urljoin(self.base_url, path.lstrip("/"))
+
+    def _request(self, method, path, *, json=None, authenticated=True):
+        headers = {"x-xsrf-token": self._token} if authenticated and self._token else {}
+        response = None
+        for attempt in range(4):
+            try:
+                response = self.session.request(method, self._url(path), json=json, headers=headers,
+                                                timeout=self.timeout, verify=self.verify_ssl)
+            except (requests.ConnectionError, requests.Timeout) as exc:
+                # DNS failures, refused connections and timeouts can be transient.
+                # Retry a few times, then report a controlled source error.
+                if attempt == 3:
+                    raise VMmanagerAPIError(
+                        f"VMmanager is unreachable for {path} after 4 attempts: {exc}"
+                    ) from exc
+                time.sleep(0.5 * (attempt + 1))
+                continue
+            except requests.RequestException as exc:
+                raise VMmanagerAPIError(f"VMmanager request failed for {path}: {exc}") from exc
+            if response.status_code != 503 or attempt == 3:
+                break
+            time.sleep(0.25 * (attempt + 1))
+        if response.status_code >= 400:
+            message = f"HTTP {response.status_code}"
+            try:
+                body = response.json()
+                error = body.get("error", {}) if isinstance(body, dict) else {}
+                if error:
+                    message += f": {error.get('msg', error)}"
+            except (ValueError, AttributeError):
+                pass
+            raise VMmanagerAPIError(f"VMmanager API {path} failed: {message}")
+        try:
+            payload = response.json()
+        except ValueError as exc:
+            raise VMmanagerAPIError(f"VMmanager API {path} returned invalid JSON") from exc
+        if isinstance(payload, dict) and payload.get("error"):
+            error = payload["error"]
+            raise VMmanagerAPIError(f"VMmanager API {path} returned error: {error.get('msg', error) if isinstance(error, dict) else error}")
+        return payload
+
+    def authenticate(self):
+        payload = self._request("POST", "auth/v4/public/token",
+                                json={"email": self.email, "password": self.password}, authenticated=False)
+        token = payload.get("token") if isinstance(payload, dict) else None
+        if not token:
+            raise VMmanagerAPIError("VMmanager authentication response did not contain a token")
+        self._token = token
+        return token
+
+    @staticmethod
+    def _extract_list(payload):
+        if isinstance(payload, list):
+            return payload
+        if not isinstance(payload, dict):
+            raise VMmanagerAPIError("VMmanager host list response is not a JSON object/list")
+        for key in ("doc", "items", "list", "data", "hosts", "host"):
+            value = payload.get(key)
+            if isinstance(value, list):
+                return value
+            if isinstance(value, dict):
+                for nested_key in ("item", "items", "host", "list"):
+                    nested = value.get(nested_key)
+                    if isinstance(nested, list):
+                        return nested
+        raise VMmanagerAPIError("Unrecognized VMmanager host list response; refusing to treat it as an empty inventory")
+
+    def _get_paged_items(self, endpoint, item_label, page_size=50):
+        """Return a complete, validated VMmanager list snapshot."""
+        if not self._token:
+            self.authenticate()
+        page_size = max(1, int(page_size))
+        offset, items, expected_total = 0, [], None
+        while True:
+            payload = self._request("GET", f"{endpoint}?limit={offset},{page_size}")
+            batch = self._extract_list(payload)
+            if any(not isinstance(item, dict) for item in batch):
+                raise VMmanagerAPIError(f"VMmanager {item_label} list contains a non-object item")
+            if any(item.get("id") is None or not str(item.get("id")).strip() for item in batch):
+                raise VMmanagerAPIError(f"VMmanager returned a {item_label} without an ID; refusing to sync")
+            if isinstance(payload, dict):
+                size = payload.get("size")
+                if isinstance(size, int) and size >= 0:
+                    expected_total = size
+            items.extend(batch)
+            offset += len(batch)
+            if expected_total is not None:
+                if len(items) >= expected_total:
+                    if len(items) != expected_total:
+                        raise VMmanagerAPIError(f"VMmanager {item_label} snapshot exceeded reported total")
+                    break
+            elif len(batch) < page_size:
+                break
+            if not batch:
+                break
+        ids = [str(item["id"]) for item in items]
+        if len(ids) != len(set(ids)):
+            raise VMmanagerAPIError(f"VMmanager returned duplicate {item_label} IDs across pages")
+        if expected_total is not None and len(items) != expected_total:
+            raise VMmanagerAPIError(f"VMmanager {item_label} snapshot incomplete: expected {expected_total}, received {len(items)}")
+        return items
+
+    def get_hosts(self, page_size=50):
+        """Return a complete VM snapshot using `/vm/v3/host?limit=offset,count`."""
+        return self._get_paged_items("vm/v3/host", "VM", page_size)
+
+    def get_disks(self, page_size=50):
+        """Return a complete disk snapshot with per-disk `size_mib` values."""
+        return self._get_paged_items("vm/v3/disk", "disk", page_size)
+
+    def get_clusters(self, page_size=50):
+        """Return a complete cluster snapshot using the VMmanager v3 API."""
+        return self._get_paged_items("vm/v3/cluster", "cluster", page_size)
+
+    def get_nodes(self, page_size=50):
+        """Return a complete hypervisor-node snapshot using the VMmanager v3 API."""
+        return self._get_paged_items("vm/v3/node", "node", page_size)

module/sources/vmmanager/config.py

@@ -0,0 +1,64 @@
+# -*- coding: utf-8 -*-
+# Licensed under the MIT license.
+from module.config import source_config_section_name
+from module.config.base import ConfigBase
+from module.config.option import ConfigOption
+from module.common.misc import quoted_split
+from module.common.logging import get_logger
+import re
+
+log = get_logger()
+
+class VMmanagerConfig(ConfigBase):
+    section_name = source_config_section_name
+    source_name_example = "my-vmmanager-example"
+
+    def __init__(self):
+        self.options = [
+            ConfigOption("enabled", bool, default_value=True, description="Enable or disable this VMmanager source."),
+            ConfigOption("type", str, default_value="vmmanager", description="Source type identifier. Must remain 'vmmanager'."),
+            ConfigOption("api_url", str, mandatory=True, description="Base URL, e.g. https://vmmanager.example.org."),
+            ConfigOption("email", str, mandatory=True, description="Email of the read-only VMmanager API account."),
+            ConfigOption("password", str, mandatory=True, sensitive=True, description="Password of the read-only VMmanager API account."),
+            ConfigOption("site_name", str, default_value="VMmanager", description="NetBox site name for imported VMs."),
+            ConfigOption("cluster_name", str, default_value=None, description="Optional NetBox cluster name; defaults to VMmanager: <source name>."),
+            ConfigOption("cluster_type_name", str, default_value="VMmanager", description="NetBox cluster type name."),
+            ConfigOption("sync_clusters", bool, default_value=True, description="Import VMmanager clusters into NetBox."),
+            ConfigOption("sync_nodes", bool, default_value=True, description="Import VMmanager hypervisor nodes as NetBox devices."),
+            ConfigOption("exclude_dummy_clusters", bool, default_value=False, description="Skip VMmanager clusters whose virtualization_type is 'dummy'."),
+            ConfigOption("hypervisor_device_type", str, default_value="VMmanager Hypervisor", description="NetBox device type model for imported hypervisor nodes."),
+            ConfigOption("hypervisor_device_role", str, default_value="Hypervisor", description="NetBox device role for imported hypervisor nodes."),
+            ConfigOption("vm_tenant_relation", str, description="Regex mapping from VM name to an existing NetBox tenant, e.g. .* = mz."),
+            ConfigOption("vm_tag_relation", str, description="Regex mapping from VM name to NetBox tag(s), e.g. .* = mz. Separate multiple mappings with commas."),
+            ConfigOption("verify_ssl", bool, default_value=True, description="Verify the VMmanager HTTPS certificate."),
+            ConfigOption("timeout", int, default_value=30, description="HTTP timeout in seconds."),
+        ]
+        super().__init__()
+
+    def validate_options(self):
+        """Parse regex relations using the same `pattern = value` syntax as VMware."""
+        for option in self.options:
+            if option.key not in ("vm_tenant_relation", "vm_tag_relation") or not option.value:
+                continue
+
+            parsed = []
+            for relation in quoted_split(option.value):
+                if "=" not in relation:
+                    log.error("Malformed relation '%s' in '%s.%s'", relation, self.source_name, option.key)
+                    self.set_validation_failed()
+                    continue
+                object_regex, assigned_name = relation.split("=", 1)
+                object_regex = object_regex.strip(' \"')
+                assigned_name = assigned_name.strip(' \"')
+                if not object_regex or not assigned_name:
+                    log.error("Malformed relation '%s' in '%s.%s'", relation, self.source_name, option.key)
+                    self.set_validation_failed()
+                    continue
+                try:
+                    compiled = re.compile(object_regex)
+                except re.error as exc:
+                    log.error("Invalid regex '%s' in '%s.%s': %s", object_regex, self.source_name, option.key, exc)
+                    self.set_validation_failed()
+                    continue
+                parsed.append({"object_regex": compiled, "assigned_name": assigned_name})
+            option.set_value(parsed)