kubevirt.core/plugins/inventory/kubevirt.py

# -*- coding: utf-8 -*-
# Copyright 2023 Red Hat, Inc.
# Based on the kubernetes.core.k8s inventory
# Apache License 2.0 (see LICENSE or http://www.apache.org/licenses/LICENSE-2.0)

from __future__ import absolute_import, division, print_function

__metaclass__ = type

DOCUMENTATION = """
name: kubevirt

short_description: Inventory source for KubeVirt VirtualMachines and VirtualMachineInstances

author:
- "KubeVirt.io Project (!UNKNOWN)"

description:
- Fetch virtual machines from one or more namespaces with an optional label selector.
- Groups by cluster name, namespaces and labels.
- Uses V(*.kubevirt.[yml|yaml]) YAML configuration file to set parameter values.
- By default it uses the active context in I(~/.kube/config) and will return all virtual machines
  for all namespaces the active user is authorized to access.

extends_documentation_fragment:
- kubevirt.core.kubevirt_auth_options
- inventory_cache
- constructed

options:
  plugin:
    description: Token that ensures this is a source file for the P(kubevirt.core.kubevirt#inventory) plugin.
    required: True
    choices: ["kubevirt", "kubevirt.core.kubevirt"]
  host_format:
    description:
    - 'Specify the format of the host in the inventory group. Available specifiers: V(name), V(namespace) and V(uid).'
    default: "{namespace}-{name}"
  name:
    description:
    - Optional name to assign to the cluster. If not provided, a name is constructed from the server
      and port.
  namespaces:
    description:
    - List of namespaces. If not specified, will fetch virtual machines from all namespaces
      the user is authorized to access.
  label_selector:
    description:
    - Define a label selector to select a subset of the fetched virtual machines.
  network_name:
    description:
    - In case multiple networks are attached to a virtual machine, define which interface should
      be returned as primary IP address.
    aliases: [ interface_name ]
  kube_secondary_dns:
    description:
    - Enable C(kubesecondarydns) derived host names when using a secondary network interface.
    type: bool
    default: False
  use_service:
    description:
    - Enable the use of C(Services) to establish an SSH connection to a virtual machine.
    - Services are only used if no O(network_name) was provided.
    type: bool
    default: True
  unset_ansible_port:
    description:
    - Try to unset the value of C(ansible_port) if no non-default value was found.
    type: bool
    default: True
    version_added: 2.2.0
  create_groups:
    description:
    - Enable the creation of groups from labels on C(VirtualMachines) and C(VirtualMachineInstances).
    type: bool
    default: False
  base_domain:
    description:
    - Override the base domain used to construct host names. Used in case of
      C(kubesecondarydns) or C(Services) of type C(NodePort) if O(append_base_domain) is set.
  append_base_domain:
    description:
    - Append the base domain of the cluster to host names constructed from SSH C(Services) of type C(NodePort).
    type: bool
    default: False
  api_version:
    description:
    - Specify the used KubeVirt API version.
    default: "kubevirt.io/v1"
  connections:
    description:
    - Optional list of cluster connection settings.
    - This parameter is deprecated. Split your connections into multiple configuration files and move
      parameters of each connection to the configuration top level.
    - Deprecated in version C(1.5.0), will be removed in version C(3.0.0).

requirements:
- "python >= 3.9"
- "kubernetes >= 28.1.0"
- "PyYAML >= 3.11"
"""

EXAMPLES = """
# Filename must end with kubevirt.[yml|yaml]

# Authenticate with token and return all virtual machines from all accessible namespaces
- plugin: kubevirt.core.kubevirt
  host: https://192.168.64.4:8443
  api_key: xxxxxxxxxxxxxxxx
  validate_certs: false

# Use default ~/.kube/config and return virtual machines from namespace testing connected to network bridge-network
- plugin: kubevirt.core.kubevirt
  namespaces:
    - testing
  network_name: bridge-network

# Use default ~/.kube/config and return virtual machines from namespace testing with label app=test
- plugin: kubevirt.core.kubevirt
  namespaces:
    - testing
  label_selector: app=test

# Use a custom config file and a specific context
- plugin: kubevirt.core.kubevirt
  kubeconfig: /path/to/config
  context: 'awx/192-168-64-4:8443/developer'
"""

from dataclasses import dataclass, InitVar
from json import loads
from re import compile as re_compile
from typing import (
    Any,
    Dict,
    List,
    Optional,
)

# Handle import errors of python kubernetes client.
# Set HAS_K8S_MODULE_HELPER and k8s_import exception accordingly to
# potentially print a warning to the user if the client is missing.
try:
    from kubernetes.dynamic.exceptions import DynamicApiError, ResourceNotFoundError

    HAS_K8S_MODULE_HELPER = True
    K8S_IMPORT_EXCEPTION = None
except ImportError as e:

    class DynamicApiError(Exception):
        """
        Dummy class, mainly used for ansible-test sanity.
        """

    class ResourceNotFoundError(Exception):
        """
        Dummy class, mainly used for ansible-test sanity.
        """

    HAS_K8S_MODULE_HELPER = False
    K8S_IMPORT_EXCEPTION = e

from ansible.plugins.inventory import BaseInventoryPlugin, Constructable, Cacheable

# Handle import errors of trust_as_template.
# It is only available on ansible-core >=2.19.
try:
    from ansible.template import trust_as_template
except ImportError:
    trust_as_template = None


from ansible_collections.kubernetes.core.plugins.module_utils.k8s.client import (
    get_api_client,
    K8SClient,
)

ANNOTATION_KUBEVIRT_IO_CLUSTER_PREFERENCE_NAME = "kubevirt.io/cluster-preference-name"
ANNOTATION_KUBEVIRT_IO_PREFERENCE_NAME = "kubevirt.io/preference-name"
ANNOTATION_VM_KUBEVIRT_IO_OS = "vm.kubevirt.io/os"
LABEL_KUBEVIRT_IO_DOMAIN = "kubevirt.io/domain"
TYPE_LOADBALANCER = "LoadBalancer"
TYPE_NODEPORT = "NodePort"
ID_MSWINDOWS = "mswindows"
SERVICE_TARGET_PORT_SSH = 22
SERVICE_TARGET_PORT_WINRM_HTTP = 5985
SERVICE_TARGET_PORT_WINRM_HTTPS = 5986


class KubeVirtInventoryException(Exception):
    """
    This class is used for exceptions raised by this inventory.
    """


@dataclass
class InventoryOptions:
    """
    This class holds the options defined by the user.
    """

    api_version: Optional[str] = None
    label_selector: Optional[str] = None
    network_name: Optional[str] = None
    kube_secondary_dns: Optional[bool] = None
    use_service: Optional[bool] = None
    unset_ansible_port: Optional[bool] = None
    create_groups: Optional[bool] = None
    base_domain: Optional[str] = None
    append_base_domain: Optional[bool] = None
    host_format: Optional[str] = None
    namespaces: Optional[List[str]] = None
    name: Optional[str] = None
    config_data: InitVar[Optional[Dict]] = None

    def __post_init__(self, config_data: Optional[Dict]) -> None:
        if not config_data or not isinstance(config_data, dict):
            config_data = {}

        # Copy values from config_data and set defaults for keys not present
        self.api_version = (
            self.api_version
            if self.api_version is not None
            else config_data.get("api_version", "kubevirt.io/v1")
        )
        self.label_selector = (
            self.label_selector
            if self.label_selector is not None
            else config_data.get("label_selector")
        )
        self.network_name = (
            self.network_name
            if self.network_name is not None
            else config_data.get("network_name", config_data.get("interface_name"))
        )
        self.kube_secondary_dns = (
            self.kube_secondary_dns
            if self.kube_secondary_dns is not None
            else config_data.get("kube_secondary_dns", False)
        )
        self.use_service = (
            self.use_service
            if self.use_service is not None
            else config_data.get("use_service", True)
        )
        self.unset_ansible_port = (
            self.unset_ansible_port
            if self.unset_ansible_port is not None
            else config_data.get("unset_ansible_port", True)
        )
        self.create_groups = (
            self.create_groups
            if self.create_groups is not None
            else config_data.get("create_groups", False)
        )
        self.base_domain = (
            self.base_domain
            if self.base_domain is not None
            else config_data.get("base_domain")
        )
        self.append_base_domain = (
            self.append_base_domain
            if self.append_base_domain is not None
            else config_data.get("append_base_domain", False)
        )
        self.host_format = (
            self.host_format
            if self.host_format is not None
            else config_data.get("host_format", "{namespace}-{name}")
        )
        self.namespaces = (
            self.namespaces
            if self.namespaces is not None
            else config_data.get("namespaces")
        )
        self.name = self.name if self.name is not None else config_data.get("name")


class InventoryModule(BaseInventoryPlugin, Constructable, Cacheable):
    """
    This class implements the actual inventory module.
    """

    NAME = "kubevirt.core.kubevirt"

    # Used to convert camel case variable names into snake case
    _snake_case_pattern = re_compile(r"(?<=[a-z])(?=[A-Z])|(?<=[A-Z])(?=[A-Z][a-z])")

    @staticmethod
    def _get_default_hostname(host: str) -> str:
        """
        _get_default_host_name strips URL schemes from the host name and
        replaces invalid characters.
        """
        return (
            host.replace("https://", "")
            .replace("http://", "")
            .replace(".", "-")
            .replace(":", "_")
        )

    @staticmethod
    def _format_dynamic_api_exc(exc: DynamicApiError) -> str:
        """
        _format_dynamic_api_exc tries to extract the message from the JSON body
        of a DynamicApiError.
        """
        if exc.body:
            if exc.headers and exc.headers.get("Content-Type") == "application/json":
                message = loads(exc.body).get("message")
                if message:
                    return message
            return exc.body

        return f"{exc.status} Reason: {exc.reason}"

    @staticmethod
    def _format_var_name(name: str) -> str:
        """
        _format_var_name formats a CamelCase variable name into a snake_case name
        suitable for use as a inventory variable name.
        """
        return InventoryModule._snake_case_pattern.sub("_", name).lower()

    @staticmethod
    def _obj_is_valid(obj: Dict) -> bool:
        """
        _obj_is_valid ensures commonly used keys are present in the passed object.
        """
        return bool(
            "spec" in obj
            and "status" in obj
            and "metadata" in obj
            and obj["metadata"].get("name")
            and obj["metadata"].get("namespace")
            and obj["metadata"].get("uid")
        )

    @staticmethod
    def _find_service_with_target_port(
        services: List[Dict], target_port: int
    ) -> Optional[Dict]:
        """
        _find_service_with_target_port returns the first found service with a given
        target port in the passed in list of services or otherwise None.
        """
        for service in services:
            if (
                (ports := service.get("spec", {}).get("ports")) is not None
                and len(ports) == 1
                and ports[0].get("targetPort", 0) == target_port
            ):
                return service

        return None

    @staticmethod
    def _get_host_from_service(
        service: Dict, node_name: Optional[str]
    ) -> Optional[str]:
        """
        _get_host_from_service extracts the hostname to be used from the
        passed in service.
        """
        service_type = service.get("spec", {}).get("type")
        if service_type == TYPE_LOADBALANCER:
            # LoadBalancer services can return a hostname or an IP address
            ingress = service.get("status", {}).get("loadBalancer", {}).get("ingress")
            if ingress is not None and len(ingress) > 0:
                hostname = ingress[0].get("hostname")
                ip_address = ingress[0].get("ip")
                return hostname if hostname is not None else ip_address
        elif service_type == TYPE_NODEPORT:
            # NodePort services use the node name as host
            return node_name

        return None

    @staticmethod
    def _get_port_from_service(service: Dict) -> Optional[str]:
        """
        _get_port_from_service extracts the port to be used from the
        passed in service.
        """
        ports = service.get("spec", {}).get("ports", [])
        if not ports:
            return None

        service_type = service.get("spec", {}).get("type")
        if service_type == TYPE_LOADBALANCER:
            # LoadBalancer services use the port attribute
            return ports[0].get("port")
        if service_type == TYPE_NODEPORT:
            # NodePort services use the nodePort attribute
            return ports[0].get("nodePort")

        return None

    @staticmethod
    def _is_windows(guest_os_info: Optional[Dict], annotations: Optional[Dict]) -> bool:
        """
        _is_windows checks whether a given VM is running a Windows guest
        by checking its GuestOSInfo and annotations.
        """
        if guest_os_info and "id" in guest_os_info:
            return guest_os_info["id"] == ID_MSWINDOWS

        if not annotations:
            return False

        if ANNOTATION_KUBEVIRT_IO_CLUSTER_PREFERENCE_NAME in annotations:
            return annotations[
                ANNOTATION_KUBEVIRT_IO_CLUSTER_PREFERENCE_NAME
            ].startswith("windows")

        if ANNOTATION_KUBEVIRT_IO_PREFERENCE_NAME in annotations:
            return annotations[ANNOTATION_KUBEVIRT_IO_PREFERENCE_NAME].startswith(
                "windows"
            )

        if ANNOTATION_VM_KUBEVIRT_IO_OS in annotations:
            return annotations[ANNOTATION_VM_KUBEVIRT_IO_OS].startswith("windows")

        return False

    def verify_file(self, path: str) -> None:
        """
        verify_file ensures the inventory file is compatible with this plugin.
        """
        return super().verify_file(path) and path.endswith(
            ("kubevirt.yml", "kubevirt.yaml")
        )

    def parse(self, inventory: Any, loader: Any, path: str, cache: bool = True) -> None:
        """
        parse is the main entry point of the inventory.
        It checks for availability of the Kubernetes Python client,
        gets the configuration, retrieves the cache or runs fetch_objects and
        populates the inventory.
        """
        if not HAS_K8S_MODULE_HELPER:
            raise KubeVirtInventoryException(
                "This module requires the Kubernetes Python client. "
                + f"Try `pip install kubernetes`. Detail: {K8S_IMPORT_EXCEPTION}"
            )

        super().parse(inventory, loader, path)

        config_data = self._read_config_data(path)
        cache_key = self.get_cache_key(path)
        user_cache_setting = self.get_option("cache")
        attempt_to_read_cache = user_cache_setting and cache
        cache_needs_update = user_cache_setting and not cache

        self._connections_compatibility(config_data)
        opts = InventoryOptions(config_data=config_data)

        results = {}
        if attempt_to_read_cache:
            try:
                results = self.cache[cache_key]
            except KeyError:
                cache_needs_update = True
        if not attempt_to_read_cache or cache_needs_update:
            results = self._fetch_objects(get_api_client(**config_data), opts)
        if cache_needs_update:
            self.cache[cache_key] = results

        self._populate_inventory(results, opts)

    def _connections_compatibility(self, config_data: Dict) -> None:
        """
        _connections_compatibility ensures compatibility with the connection
        parameter found in earlier versions of this inventory plugin (<1.5.0).
        """
        collection_name = "kubevirt.core"
        version_removed_in = "3.0.0"

        if (connections := config_data.get("connections")) is None:
            return

        self.display.deprecated(
            msg="The 'connections' parameter is deprecated and now supports only a single list entry.",
            version=version_removed_in,
            collection_name=collection_name,
        )

        if not isinstance(connections, list):
            raise KubeVirtInventoryException("Expecting connections to be a list.")

        if len(connections) == 1:
            if not isinstance(connections[0], dict):
                raise KubeVirtInventoryException(
                    "Expecting connection to be a dictionary."
                )
            # Copy the single connections entry into the top level
            for k, v in connections[0].items():
                config_data[k] = v
            self.display.deprecated(
                msg="Move all of your connection parameters to the configuration top level.",
                version=version_removed_in,
                collection_name=collection_name,
            )
        elif len(connections) > 1:
            self.display.deprecated(
                msg="Split your connections into multiple configuration files.",
                version="2.0.0",
                collection_name=collection_name,
                removed=True,
            )

    def _fetch_objects(self, client: Any, opts: InventoryOptions) -> Dict:
        """
        fetch_objects fetches all relevant objects from the K8S API.
        """
        namespaces = {}
        for namespace in (
            opts.namespaces
            if opts.namespaces
            else self._get_available_namespaces(client)
        ):
            vms = self._get_vms_for_namespace(client, namespace, opts)
            vmis = self._get_vmis_for_namespace(client, namespace, opts)

            if not vms and not vmis:
                # Continue if no VMs and VMIs were found to avoid adding empty groups.
                continue

            namespaces[namespace] = {
                "vms": vms,
                "vmis": vmis,
                "services": self._get_services_for_namespace(client, namespace),
            }

        return {
            "default_hostname": self._get_default_hostname(client.configuration.host),
            "cluster_domain": self._get_cluster_domain(client),
            "namespaces": namespaces,
        }

    def _get_cluster_domain(self, client: K8SClient) -> Optional[str]:
        """
        _get_cluster_domain tries to get the base domain of an OpenShift cluster.
        """
        try:
            v1_dns = client.resources.get(
                api_version="config.openshift.io/v1", kind="DNS"
            )
        except Exception:
            # If resource not found return None
            return None
        try:
            obj = v1_dns.get(name="cluster")
        except DynamicApiError as exc:
            self.display.debug(
                f"Failed to fetch cluster DNS config: {self._format_dynamic_api_exc(exc)}"
            )
            return None
        return obj.get("spec", {}).get("baseDomain")

    def _get_resources(
        self, client: K8SClient, api_version: str, kind: str, **kwargs
    ) -> List[Dict]:
        """
        _get_resources uses a dynamic K8SClient to fetch resources from the K8S API.
        """
        client = client.resources.get(api_version=api_version, kind=kind)
        try:
            result = client.get(**kwargs)
        except DynamicApiError as exc:
            self.display.debug(exc)
            raise KubeVirtInventoryException(
                f"Error fetching {kind} list: {self._format_dynamic_api_exc(exc)}"
            ) from exc

        return [item.to_dict() for item in result.items]

    def _get_available_namespaces(self, client: K8SClient) -> List[str]:
        """
        _get_available_namespaces lists all namespaces accessible with the
        configured credentials and returns them.
        """

        namespaces = []
        try:
            namespaces = self._get_resources(
                client, "project.openshift.io/v1", "Project"
            )
        except ResourceNotFoundError:
            namespaces = self._get_resources(client, "v1", "Namespace")

        return [
            namespace["metadata"]["name"]
            for namespace in namespaces
            if "metadata" in namespace and "name" in namespace["metadata"]
        ]

    def _get_vms_for_namespace(
        self, client: K8SClient, namespace: str, opts: InventoryOptions
    ) -> List[Dict]:
        """
        _get_vms_for_namespace returns a list of all VirtualMachines in a namespace.
        """
        return self._get_resources(
            client,
            opts.api_version,
            "VirtualMachine",
            namespace=namespace,
            label_selector=opts.label_selector,
        )

    def _get_vmis_for_namespace(
        self, client: K8SClient, namespace: str, opts: InventoryOptions
    ) -> List[Dict]:
        """
        _get_vmis_for_namespace returns a list of all VirtualMachineInstances in a namespace.
        """
        return self._get_resources(
            client,
            opts.api_version,
            "VirtualMachineInstance",
            namespace=namespace,
            label_selector=opts.label_selector,
        )

    def _get_services_for_namespace(
        self, client: K8SClient, namespace: str
    ) -> Dict[str, List[Dict]]:
        """
        _get_services_for_namespace retrieves all services of a namespace exposing ssh or winrm.
        The services are mapped to the name of the corresponding domain.
        """
        items = self._get_resources(
            client,
            "v1",
            "Service",
            namespace=namespace,
        )

        services = {}
        for service in items:
            # Continue if service is not of type LoadBalancer or NodePort
            if not (spec := service.get("spec")):
                continue

            if spec.get("type") not in (
                TYPE_LOADBALANCER,
                TYPE_NODEPORT,
            ):
                continue

            # Continue if ports are not defined, there are more than one port mapping
            # or the target port is not port 22 (ssh) or port 5985 or 5986 (winrm).
            if (
                (ports := spec.get("ports")) is None
                or len(ports) != 1
                or ports[0].get("targetPort")
                not in [
                    SERVICE_TARGET_PORT_SSH,
                    SERVICE_TARGET_PORT_WINRM_HTTP,
                    SERVICE_TARGET_PORT_WINRM_HTTPS,
                ]
            ):
                continue

            # Only add the service to the list if the domain selector is present
            if domain := spec.get("selector", {}).get(LABEL_KUBEVIRT_IO_DOMAIN):
                if domain not in services:
                    services[domain] = []
                services[domain].append(service)

        return services

    def _populate_inventory(self, results: Dict, opts: InventoryOptions) -> None:
        """
        _populate_inventory populates the inventory by completing the InventoryOptions
        and invoking populate_inventory_from_namespace for every namespace in results.
        """
        if opts.base_domain is None:
            opts.base_domain = results["cluster_domain"]
        if opts.name is None:
            opts.name = results["default_hostname"]
        for namespace, data in results["namespaces"].items():
            self._populate_inventory_from_namespace(namespace, data, opts)

    def _populate_inventory_from_namespace(
        self, namespace: str, data: Dict, opts: InventoryOptions
    ) -> None:
        """
        _populate_inventory_from_namespace adds groups and hosts from a
        namespace to the inventory.
        """
        vms = {
            vm["metadata"]["name"]: vm for vm in data["vms"] if self._obj_is_valid(vm)
        }
        vmis = {
            vmi["metadata"]["name"]: vmi
            for vmi in data["vmis"]
            if self._obj_is_valid(vmi)
        }

        if not vms and not vmis:
            # Return early if no VMs and VMIs were found to avoid adding empty groups.
            return

        services = {
            domain: [service for service in services if self._obj_is_valid(service)]
            for domain, services in data["services"].items()
        }

        name = self._sanitize_group_name(opts.name)
        namespace_group = self._sanitize_group_name(f"namespace_{namespace}")

        self.inventory.add_group(name)
        self.inventory.add_group(namespace_group)
        self.inventory.add_child(name, namespace_group)

        # Add found VMs and optionally enhance with VMI data
        for name, vm in vms.items():
            hostname = self._add_host(vm["metadata"], opts.host_format, namespace_group)
            self._set_vars_from_vm(hostname, vm, opts)
            if name in vmis:
                self._set_vars_from_vmi(hostname, vmis[name], services, opts)
            self._set_composable_vars(hostname)

        # Add remaining VMIs without VM
        for name, vmi in vmis.items():
            if name in vms:
                continue
            hostname = self._add_host(
                vmi["metadata"], opts.host_format, namespace_group
            )
            self._set_vars_from_vmi(hostname, vmi, services, opts)
            self._set_composable_vars(hostname)

    def _add_host(self, metadata: Dict, host_format: str, namespace_group: str) -> str:
        """
        _add_host adds a host to the inventory.
        """
        hostname = host_format.format(
            namespace=metadata["namespace"],
            name=metadata["name"],
            uid=metadata["uid"],
        )
        self.inventory.add_host(hostname)
        self.inventory.add_child(namespace_group, hostname)

        return hostname

    def _set_vars_from_vm(
        self, hostname: str, vm: Dict, opts: InventoryOptions
    ) -> None:
        """
        _set_vars_from_vm sets inventory variables from a VM prefixed with vm_.
        """
        self._set_common_vars(hostname, "vm", vm, opts)

    def _set_vars_from_vmi(
        self,
        hostname: str,
        vmi: Dict,
        services: Dict[str, List[Dict]],
        opts: InventoryOptions,
    ) -> None:
        """
        _set_vars_from_vmi sets inventory variables from a VMI prefixed with vmi_ and
        looks up the interface to set ansible_host and ansible_port.
        """
        self._set_common_vars(hostname, "vmi", vmi, opts)

        if not (interfaces := vmi["status"].get("interfaces")):
            return

        if opts.network_name is None:
            # Use first interface
            interface = interfaces[0]
        else:
            # Find interface by its name
            interface = next(
                (i for i in interfaces if i.get("name") == opts.network_name),
                None,
            )

        # If interface is not found or IP address is not reported skip this VMI
        if not interface or not interface.get("ipAddress"):
            return

        _services = services.get(
            vmi["metadata"].get("labels", {}).get(LABEL_KUBEVIRT_IO_DOMAIN), []
        )

        # Set up the connection
        service = None
        if self._is_windows(
            vmi["status"].get("guestOSInfo", {}),
            vmi["metadata"].get("annotations", {}),
        ):
            self.inventory.set_variable(hostname, "ansible_connection", "winrm")
            service = self._find_service_with_target_port(
                _services, SERVICE_TARGET_PORT_WINRM_HTTPS
            )
            if service is None:
                service = self._find_service_with_target_port(
                    _services, SERVICE_TARGET_PORT_WINRM_HTTP
                )
        else:
            service = self._find_service_with_target_port(
                _services, SERVICE_TARGET_PORT_SSH
            )

        self._set_ansible_host_and_port(
            vmi,
            hostname,
            interface["ipAddress"],
            service,
            opts,
        )

    def _set_common_vars(
        self, hostname: str, prefix: str, obj: Dict, opts: InventoryOptions
    ):
        """
        _set_common_vars sets common inventory variables from VMs or VMIs.
        """
        # Add hostvars from metadata
        if annotations := obj["metadata"].get("annotations"):
            self.inventory.set_variable(hostname, f"{prefix}_annotations", annotations)
        if labels := obj["metadata"].get("labels"):
            self.inventory.set_variable(hostname, f"{prefix}_labels", labels)
            # Create label groups and add vm to it if enabled
            if opts.create_groups:
                self._set_groups_from_labels(hostname, labels)
        if resource_version := obj["metadata"].get("resourceVersion"):
            self.inventory.set_variable(
                hostname, f"{prefix}_resource_version", resource_version
            )
        if uid := obj["metadata"].get("uid"):
            self.inventory.set_variable(hostname, f"{prefix}_uid", uid)

        # Add hostvars from status
        for key, value in obj["status"].items():
            self.inventory.set_variable(
                hostname, f"{prefix}_{self._format_var_name(key)}", value
            )

    def _set_groups_from_labels(self, hostname: str, labels: Dict) -> None:
        """
        _set_groups_from_labels adds groups for each label of a VM or VMI and
        adds the host to each group.
        """
        groups = []
        for key, value in labels.items():
            group_name = self._sanitize_group_name(f"label_{key}_{value}")
            if group_name not in groups:
                groups.append(group_name)
        # Add host to each label_value group
        for group in groups:
            self.inventory.add_group(group)
            self.inventory.add_child(group, hostname)

    def _set_ansible_host_and_port(
        self,
        vmi: Dict,
        hostname: str,
        ip_address: str,
        service: Optional[Dict],
        opts: InventoryOptions,
    ) -> None:
        """
        _set_ansible_host_and_port sets the ansible_host and possibly the ansible_port var.
        Secondary interfaces have priority over a service exposing SSH
        """
        ansible_host = None
        ansible_port = None
        if opts.kube_secondary_dns and opts.network_name:
            # Set ansible_host to the kubesecondarydns derived host name if enabled
            # See https://github.com/kubevirt/kubesecondarydns#parameters
            ansible_host = f"{opts.network_name}.{vmi['metadata']['name']}.{vmi['metadata']['namespace']}.vm"
            if opts.base_domain:
                ansible_host += f".{opts.base_domain}"
        elif opts.use_service and service and not opts.network_name:
            # Set ansible_host and ansible_port to the host and port from the LoadBalancer
            # or NodePort service exposing SSH
            node_name = vmi["status"].get("nodeName")
            if node_name and opts.append_base_domain and opts.base_domain:
                node_name += f".{opts.base_domain}"
            host = self._get_host_from_service(service, node_name)
            port = self._get_port_from_service(service)
            if host is not None and port is not None:
                ansible_host = host
                ansible_port = port

        # Default to the IP address of the interface if ansible_host was not set prior
        if ansible_host is None:
            ansible_host = ip_address

        self.inventory.set_variable(hostname, "ansible_host", ansible_host)
        if opts.unset_ansible_port or ansible_port is not None:
            self.inventory.set_variable(hostname, "ansible_port", ansible_port)

    def _set_composable_vars(self, hostname: str) -> None:
        """
        _set_composable_vars sets vars per
        https://docs.ansible.com/ansible/latest/dev_guide/developing_inventory.html
        """
        hostvars = self.inventory.get_host(hostname).get_vars()
        strict = self.get_option("strict")

        def trust_compose_groups(data: Dict) -> Dict:
            if trust_as_template is not None:
                return {k: trust_as_template(v) for k, v in data.items()}
            return data

        def trust_keyed_groups(data: List) -> List:
            if trust_as_template is not None:
                return [{**d, "key": trust_as_template(d["key"])} for d in data]
            return data

        self._set_composite_vars(
            trust_compose_groups(self.get_option("compose")),
            hostvars,
            hostname,
            strict=True,
        )
        self._add_host_to_composed_groups(
            trust_compose_groups(self.get_option("groups")),
            hostvars,
            hostname,
            strict=strict,
        )
        self._add_host_to_keyed_groups(
            trust_keyed_groups(self.get_option("keyed_groups")),
            hostvars,
            hostname,
            strict=strict,
        )