Add Incus inventory plugin (#10972)

* BOTMETA: Add Incus inventory plugin

Signed-off-by: Stéphane Graber <stgraber@stgraber.org>

* plugins/inventory: Implement basic Incus support

This is a simple inventory plugin leveraging the local `incus` command
line tool. It supports accessing multiple remotes and projects, builds a
simple group hierarchy based on the remotes and projects and exposes
most properties as variable. It also supports basic filtering using the
server-side filtering syntax supported by the Incus CLI.

Signed-off-by: Stéphane Graber <stgraber@stgraber.org>

* plugins/inventory/incus: Add support for constructable groups

This allows the use of constructable groups and also allows disabling
the default group structure.

Signed-off-by: Stéphane Graber <stgraber@stgraber.org>

* plugins/inventory/incus: Add unit tests

Signed-off-by: Stéphane Graber <stgraber@stgraber.org>

---------

Signed-off-by: Stéphane Graber <stgraber@stgraber.org>

.github/BOTMETA.yml

@@ -238,6 +238,9 @@ files:
     maintainers: vbotka
   $inventories/icinga2.py:
     maintainers: BongoEADGC6
+  $inventories/incus.py:
+    labels: incus
+    maintainers: stgraber
   $inventories/linode.py:
     keywords: linode dynamic inventory script
     labels: cloud linode

plugins/inventory/incus.py

@@ -0,0 +1,232 @@
+# Copyright (c) 2025 Stéphane Graber <stgraber@stgraber.org>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from __future__ import annotations
+
+DOCUMENTATION = r"""
+name: incus
+short_description: Incus inventory source
+version_added: 12.0.0
+author:
+  - Stéphane Graber (@stgraber)
+requirements:
+  - Incus CLI (C(incus))
+description:
+  - Get inventory hosts from the Incus container and virtual-machine manager.
+options:
+  plugin:
+    description:
+      - The name of this plugin, it should always be set to community.general.incus for this plugin to work.
+    required: true
+    choices: ['community.general.incus']
+    type: str
+  default_groups:
+    description:
+      - Whether to generate default groups based on remote and project.
+    type: bool
+    default: true
+  filters:
+    description:
+      - Filter expression as supported by C(incus list).
+    type: list
+    elements: string
+    default: []
+  host_domain:
+    description:
+      - Domain to append to the host FQDN.
+    type: string
+  host_fqdn:
+    description:
+      - Whether to generate a FQDN for the host name.
+      - This will use the INSTANCE.PROJECT.REMOTE syntax.
+    type: bool
+    default: true
+  remotes:
+    description:
+      - The names of the Incus remotes to use (per C(incus remote list)).
+      - Remotes are used to access multiple servers from a single client.
+      - By default the inventory will go over all projects for each remote.
+      - It is possible to specify a specific project using V(remote:project).
+    type: list
+    elements: string
+    default: ["local"]
+extends_documentation_fragment:
+  - ansible.builtin.constructed
+"""
+
+EXAMPLES = r"""
+---
+# Pull instances from all projects on the local remote.
+plugin: community.general.incus
+
+---
+# Pull running VMs from all projects on the local remote.
+plugin: community.general.incus
+filters:
+  - type=virtual-machine
+  - status=running
+
+---
+# Pull instances from two different remotes
+plugin: community.general.incus
+remotes:
+  - remote-1
+  - remote-2
+
+---
+# Pull instances from two different remotes
+# Limiting the second to the default project
+plugin: community.general.incus
+remotes:
+  - remote-1
+  - remote-2:default
+"""
+
+from ansible.plugins.inventory import BaseInventoryPlugin, Constructable
+from ansible.utils.display import Display
+from json import loads
+from subprocess import check_output
+
+display = Display()
+
+
+class InventoryModule(BaseInventoryPlugin, Constructable):
+    """Host inventory parser for Incus."""
+
+    NAME = "community.general.incus"
+
+    def __init__(self):
+        super(InventoryModule, self).__init__()
+
+    def verify_file(self, path):
+        valid = False
+
+        if super(InventoryModule, self).verify_file(path):
+            if path.endswith(("incus.yaml", "incus.yml")):
+                valid = True
+            else:
+                self.display.vvv(
+                    'Skipping due to inventory source not ending in "incus.yaml" nor "incus.yml"'
+                )
+
+        return valid
+
+    def parse(self, inventory, loader, path, cache=True):
+        super(InventoryModule, self).parse(inventory, loader, path)
+
+        self._read_config_data(path)
+
+        self.populate()
+
+    def populate(self):
+        # Create top-level "incus" group if missing.
+        default_groups = self.get_option("default_groups")
+        if default_groups:
+            self.inventory.add_group("incus")
+
+        for remote in self.get_option("remotes"):
+            # Split the remote name from the project name (if specified).
+            remote_name = ""
+            project_name = ""
+
+            fields = remote.split(":", 1)
+            if len(fields) == 2:
+                remote_name = fields[0]
+                project_name = fields[1]
+            else:
+                remote_name = fields[0]
+
+            # Create the remote-specific group if missing.
+            group_remote = f"incus_{remote_name}"
+            if default_groups:
+                self.inventory.add_group(group_remote)
+                self.inventory.add_child("incus", group_remote)
+
+            # Get a list of projects.
+            projects = []
+            if project_name:
+                projects = [project_name]
+            else:
+                projects = [
+                    entry["name"]
+                    for entry in self._run_incus("project", "list", f"{remote_name}:")
+                ]
+
+            # Get a list of instances.
+            for project in projects:
+                # Create the project-specific group if missing.
+                group_project = f"{group_remote}_{project}"
+                if default_groups:
+                    self.inventory.add_group(group_project)
+                    self.inventory.add_child(group_remote, group_project)
+
+                # List the instances.
+                list_cmd = [
+                    "list",
+                    f"{remote_name}:",
+                    "--project",
+                    project,
+                ] + self.get_option("filters")
+                for instance in self._run_incus(*list_cmd):
+                    # Compute the host name.
+                    host_name = instance["name"]
+                    if self.get_option("host_fqdn"):
+                        host_name = f"{host_name}.{project}.{remote_name}"
+
+                        domain = self.get_option("host_domain")
+                        if domain:
+                            host_name = f"{host_name}.{domain}"
+
+                    # Add some extra variables.
+                    host_vars = {}
+                    host_vars["ansible_incus_remote"] = remote_name
+                    host_vars["ansible_incus_project"] = project
+
+                    for prop in (
+                        "architecture",
+                        "config",
+                        "description",
+                        "devices",
+                        "ephemeral",
+                        "expanded_config",
+                        "expanded_devices",
+                        "location",
+                        "profiles",
+                        "status",
+                        "type",
+                    ):
+                        host_vars[f"ansible_incus_{prop}"] = instance[prop]
+
+                    # Add the host to the inventory and constructed groups.
+                    self._add_host(host_name, host_vars)
+
+                    # Add the host to the built-in groups.
+                    if default_groups:
+                        self.inventory.add_host(host_name, group_project)
+
+    def _add_host(self, hostname, host_vars):
+        self.inventory.add_host(hostname, group="all")
+
+        for var_name, var_value in host_vars.items():
+            self.inventory.set_variable(hostname, var_name, var_value)
+
+        strict = self.get_option("strict")
+
+        # Add variables created by the user's Jinja2 expressions to the host
+        self._set_composite_vars(
+            self.get_option("compose"), host_vars, hostname, strict=True
+        )
+
+        # Create user-defined groups using variables and Jinja2 conditionals
+        self._add_host_to_composed_groups(
+            self.get_option("groups"), host_vars, hostname, strict=strict
+        )
+        self._add_host_to_keyed_groups(
+            self.get_option("keyed_groups"), host_vars, hostname, strict=strict
+        )
+
+    def _run_incus(self, *args):
+        local_cmd = ["incus"] + list(args) + ["--format=json"]
+        stdout = check_output(local_cmd)
+        return loads(stdout)

tests/unit/plugins/inventory/test_incus.py

@@ -0,0 +1,143 @@
+# Copyright (c) 2025 Stéphane Graber <stgraber@stgraber.org>
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+from __future__ import annotations
+
+import pytest
+
+from ansible.inventory.data import InventoryData
+from ansible.parsing.dataloader import DataLoader
+from ansible.template import Templar
+from ansible_collections.community.general.plugins.inventory.incus import (
+    InventoryModule,
+)
+
+
+@pytest.fixture(scope="module")
+def inventory():
+    plugin = InventoryModule()
+    plugin.inventory = InventoryData()
+    plugin.templar = Templar(loader=DataLoader())
+    return plugin
+
+
+def test_verify_file_yml(tmp_path, inventory):
+    file = tmp_path / "foobar.incus.yml"
+    file.touch()
+    assert inventory.verify_file(str(file)) is True
+
+
+def test_verify_file_yaml(tmp_path, inventory):
+    file = tmp_path / "foobar.incus.yaml"
+    file.touch()
+    assert inventory.verify_file(str(file)) is True
+
+
+def test_verify_file_bad_config_yml(inventory):
+    assert inventory.verify_file("foobar.incus.yml") is False
+
+
+def test_verify_file_bad_config_yaml(inventory):
+    assert inventory.verify_file("foobar.incus.yaml") is False
+
+
+def test_verify_file_bad_config(inventory):
+    assert inventory.verify_file("foobar.wrongcloud.yml") is False
+
+
+def get_option(option):
+    if option == "default_groups":
+        return True
+
+    if option == "remotes":
+        return ["r1", "r2", "r3:proj1", "r3:proj2"]
+
+    if option == "filters":
+        return ["status=running"]
+
+    if option == "host_fqdn":
+        return True
+
+    if option == "host_domain":
+        return "example.net"
+
+    return False
+
+
+def _make_host(name):
+    entry = {}
+    entry["name"] = name
+
+    for prop in (
+        "architecture",
+        "config",
+        "description",
+        "devices",
+        "ephemeral",
+        "expanded_config",
+        "expanded_devices",
+        "location",
+        "profiles",
+        "status",
+        "type",
+    ):
+        entry[prop] = ""
+
+    return entry
+
+
+def run_incus(*args):
+    if args == ("project", "list", "r1:"):
+        return [{"name": "default"}]
+
+    if args == ("project", "list", "r2:"):
+        return [{"name": "foo"}]
+
+    if args == ("list", "r1:", "--project", "default", "status=running"):
+        return [_make_host("c1")]
+
+    if args == ("list", "r2:", "--project", "foo", "status=running"):
+        return [_make_host("c2")]
+
+    if args == ("list", "r3:", "--project", "proj1", "status=running"):
+        return [_make_host("c3")]
+
+    if args == ("list", "r3:", "--project", "proj2", "status=running"):
+        return [_make_host("c4"), _make_host("c5")]
+
+    return []
+
+
+def test_build_inventory(inventory, mocker):
+    inventory.get_option = mocker.MagicMock(side_effect=get_option)
+    inventory._run_incus = mocker.MagicMock(side_effect=run_incus)
+    inventory.populate()
+
+    c1 = inventory.inventory.get_host("c1.default.r1.example.net")
+    assert c1
+    assert "ansible_incus_status" in c1.get_vars()
+
+    c2 = inventory.inventory.get_host("c2.foo.r2.example.net")
+    assert c2
+    assert "ansible_incus_status" in c2.get_vars()
+
+    c3 = inventory.inventory.get_host("c3.proj1.r3.example.net")
+    assert c3
+    assert "ansible_incus_status" in c3.get_vars()
+
+    c4 = inventory.inventory.get_host("c4.proj2.r3.example.net")
+    assert c4
+    assert "ansible_incus_status" in c4.get_vars()
+
+    c5 = inventory.inventory.get_host("c5.proj2.r3.example.net")
+    assert c5
+    assert "ansible_incus_status" in c5.get_vars()
+
+    assert len(inventory.inventory.groups["all"].hosts) == 5
+    assert len(inventory.inventory.groups["incus"].child_groups) == 3
+    assert len(inventory.inventory.groups["incus_r1"].child_groups) == 1
+    assert len(inventory.inventory.groups["incus_r2"].child_groups) == 1
+    assert len(inventory.inventory.groups["incus_r3"].child_groups) == 2
+    assert len(inventory.inventory.groups["incus_r3_proj1"].hosts) == 1
+    assert len(inventory.inventory.groups["incus_r3_proj2"].hosts) == 2