Merge pull request #1106 from neXenio/feat/inventory-interface

feat: inventory interface for pluggable inventory

kapitan/cli.py

@@ -102,6 +102,15 @@ def build_parser():
     parser.add_argument("--version", action="version", version=VERSION)
     subparser = parser.add_subparsers(help="commands", dest="subparser_name")
 
+    inventory_backend_parser = argparse.ArgumentParser(add_help=False)
+    inventory_backend_group = inventory_backend_parser.add_argument_group("inventory_backend")
+    inventory_backend_group.add_argument(
+        "--reclass",
+        action="store_true",
+        default=from_dot_kapitan("inventory_backend", "reclass", False),
+        help="use reclass as inventory backend (default)",
+    )
+
     eval_parser = subparser.add_parser("eval", aliases=["e"], help="evaluate jsonnet file")
     eval_parser.add_argument("jsonnet_file", type=str)
     eval_parser.set_defaults(func=trigger_eval, name="eval")
@@ -131,7 +140,9 @@ def build_parser():
         help='set search paths, default is ["."]',
     )
 
-    compile_parser = subparser.add_parser("compile", aliases=["c"], help="compile targets")
+    compile_parser = subparser.add_parser(
+        "compile", aliases=["c"], help="compile targets", parents=[inventory_backend_parser]
+    )
     compile_parser.set_defaults(func=trigger_compile, name="compile")
 
     compile_parser.add_argument(
@@ -326,7 +337,9 @@ def build_parser():
         metavar="key=value",
     )
 
-    inventory_parser = subparser.add_parser("inventory", aliases=["i"], help="show inventory")
+    inventory_parser = subparser.add_parser(
+        "inventory", aliases=["i"], help="show inventory", parents=[inventory_backend_parser]
+    )
     inventory_parser.set_defaults(func=generate_inventory, name="inventory")
 
     inventory_parser.add_argument(
@@ -414,7 +427,9 @@ def build_parser():
     secrets_parser = subparser.add_parser("secrets", aliases=["s"], help="(DEPRECATED) please use refs")
     secrets_parser.set_defaults(func=print_deprecated_secrets_msg, name="secrets")
 
-    refs_parser = subparser.add_parser("refs", aliases=["r"], help="manage refs")
+    refs_parser = subparser.add_parser(
+        "refs", aliases=["r"], help="manage refs", parents=[inventory_backend_parser]
+    )
     refs_parser.set_defaults(func=handle_refs_command, name="refs")
 
     refs_parser.add_argument(

kapitan/inputs/jsonnet.py

@@ -10,6 +10,7 @@
 import os
 import sys
 
+from kapitan import cached
 from kapitan.errors import CompileError
 from kapitan.inputs.base import CompiledFile, InputType
 from kapitan.resources import resource_callbacks, search_imports

kapitan/inventory/__init__.py

@@ -0,0 +1,2 @@
+from .inv_reclass import ReclassInventory
+from .inventory import Inventory

kapitan/inventory/inv_reclass.py

@@ -0,0 +1,93 @@
+import logging
+import os
+
+import reclass
+import reclass.core
+import yaml
+from reclass.errors import NotFoundError, ReclassException
+
+from kapitan.errors import InventoryError
+
+from .inventory import Inventory
+
+logger = logging.getLogger(__name__)
+
+
+class ReclassInventory(Inventory):
+
+    def render_targets(self, targets: list = None, ignore_class_notfound: bool = False):
+        """
+        Runs a reclass inventory in inventory_path
+        (same output as running ./reclass.py -b inv_base_uri/ --inventory)
+        Will attempt to read reclass config from 'reclass-config.yml' otherwise
+        it will fall back to the default config.
+        Returns a reclass style dictionary
+
+        Does not throw errors if a class is not found while ignore_class_notfound is specified
+        """
+        reclass_config = get_reclass_config(self.inventory_path)
+        reclass_config.setdefault("ignore_class_notfound", ignore_class_notfound)
+        reclass_config["compose_node_name"] = self.compose_target_name
+
+        try:
+            storage = reclass.get_storage(
+                reclass_config["storage_type"],
+                reclass_config["nodes_uri"],
+                reclass_config["classes_uri"],
+                reclass_config["compose_node_name"],
+            )
+            class_mappings = reclass_config.get("class_mappings")  # this defaults to None (disabled)
+            _reclass = reclass.core.Core(storage, class_mappings, reclass.settings.Settings(reclass_config))
+            rendered_inventory = _reclass.inventory()
+
+            # store parameters and classes
+            for target_name, rendered_target in rendered_inventory["nodes"].items():
+                self.targets[target_name].parameters = rendered_target["parameters"]
+
+            for class_name, referenced_targets in rendered_inventory["classes"].items():
+                for target_name in referenced_targets:
+                    self.targets[target_name].classes += class_name
+
+        except ReclassException as e:
+            if isinstance(e, NotFoundError):
+                logger.error("Inventory reclass error: inventory not found")
+            else:
+                logger.error(f"Inventory reclass error: {e.message}")
+            raise InventoryError(e.message)
+
+
+def get_reclass_config(inventory_path: str) -> dict:
+    # set default values initially
+    reclass_config = {
+        "storage_type": "yaml_fs",
+        "inventory_base_uri": inventory_path,
+        "nodes_uri": "targets",
+        "classes_uri": "classes",
+        "compose_node_name": False,
+        "allow_none_override": True,
+    }
+    try:
+        from yaml import CSafeLoader as YamlLoader
+    except ImportError:
+        from yaml import SafeLoader as YamlLoader
+
+    # get reclass config from file 'inventory/reclass-config.yml'
+    cfg_file = os.path.join(inventory_path, "reclass-config.yml")
+    if os.path.isfile(cfg_file):
+        with open(cfg_file, "r") as fp:
+            config = yaml.load(fp.read(), Loader=YamlLoader)
+            logger.debug(f"Using reclass inventory config at: {cfg_file}")
+        if config:
+            # set attributes, take default values if not present
+            for key, value in config.items():
+                reclass_config[key] = value
+        else:
+            logger.debug(f"Reclass config: Empty config file at {cfg_file}. Using reclass inventory config defaults")
+    else:
+        logger.debug("Inventory reclass: No config file found. Using reclass inventory config defaults")
+
+    # normalise relative nodes_uri and classes_uri paths
+    for uri in ("nodes_uri", "classes_uri"):
+        reclass_config[uri] = os.path.normpath(os.path.join(inventory_path, reclass_config[uri]))
+
+    return reclass_config

kapitan/inventory/inventory.py

@@ -0,0 +1,154 @@
+#!/usr/bin/env python3
+
+# Copyright 2023 The Kapitan Authors
+# SPDX-FileCopyrightText: 2023 The Kapitan Authors <[email protected]>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+import logging
+import os
+import time
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import overload, Union
+
+from kapitan.errors import KapitanError
+from kapitan.reclass.reclass.values import item
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class InventoryTarget:
+    name: str
+    path: str
+    composed_name: str
+    parameters: dict = field(default_factory=dict)
+    classes: list = field(default_factory=list)
+
+
+class Inventory(ABC):
+    _default_path: str = "inventory"
+
+    def __init__(self, inventory_path: str = _default_path, compose_target_name: bool = False):
+        self.inventory_path = inventory_path
+        self.targets_path = os.path.join(inventory_path, "targets")
+        self.classes_path = os.path.join(inventory_path, "classes")
+
+        # config
+        self.compose_target_name = compose_target_name
+
+        self.targets = {}
+
+    @property
+    def inventory(self) -> dict:
+        """
+        get all targets from inventory
+        targets will be rendered
+        """
+        if not self.targets:
+            self.search_targets()
+
+        inventory = self.get_targets([*self.targets.keys()])
+
+        return {
+            target_name: {"parameters": target.parameters, "classes": target.classes}
+            for target_name, target in inventory.items()
+        }
+
+    def search_targets(self) -> dict:
+        """
+        look for targets at '<inventory_path>/targets/' and return targets without rendering parameters
+        """
+        for root, dirs, files in os.walk(self.targets_path):
+            for file in files:
+                # split file extension and check if yml/yaml
+                path = os.path.join(root, file)
+                name, ext = os.path.splitext(file)
+                if ext not in (".yml", ".yaml"):
+                    logger.debug(f"{file}: targets have to be .yml or .yaml files.")
+                    continue
+
+                # initialize target
+                composed_name = (
+                    os.path.splitext(os.path.relpath(path, self.targets_path))[0]
+                    .replace(os.sep, ".")
+                    .lstrip(".")
+                )
+                target = InventoryTarget(name, path, composed_name)
+                if self.compose_target_name:
+                    target.name = target.composed_name
+
+                # check for same name
+                if self.targets.get(target.name):
+                    raise InventoryError(
+                        f"Conflicting targets {target.name}: {target.path} and {self.targets[target.name].path}"
+                    )
+
+                self.targets[target.name] = target
+
+        return self.targets
+
+    def get_target(self, target_name: str, ignore_class_not_found: bool = False) -> InventoryTarget:
+        """
+        helper function to get rendered InventoryTarget object for single target
+        """
+        return self.get_targets([target_name], ignore_class_not_found)[target_name]
+
+    def get_targets(self, target_names: list, ignore_class_not_found: bool = False) -> dict:
+        """
+        helper function to get rendered InventoryTarget objects for multiple targets
+        """
+        targets_to_render = []
+
+        for target_name in target_names:
+            target = self.targets.get(target_name)
+            if not target:
+                if ignore_class_not_found:
+                    continue
+                raise InventoryError(f"target '{target_name}' not found")
+
+            if not target.parameters:
+                targets_to_render.append(target)
+
+        self.render_targets(targets_to_render, ignore_class_not_found)
+
+        return {name: target for name, target in self.targets.items() if name in target_names}
+
+    def get_parameters(self, target_names: Union[str, list], ignore_class_not_found: bool = False) -> dict:
+        """
+        helper function to get rendered parameters for single target or multiple targets
+        """
+        if type(target_names) is str:
+            target = self.get_target(target_names, ignore_class_not_found)
+            return target.parameters
+
+        return {name: target.parameters for name, target in self.get_targets(target_names)}
+
+    @abstractmethod
+    def render_targets(self, targets: list = None, ignore_class_notfound: bool = False):
+        """
+        create the inventory depending on which backend gets used
+        """
+        raise NotImplementedError
+
+    def __getitem__(self, key):
+        return self.inventory[key]
+
+
+class InventoryError(KapitanError):
+    """inventory error"""
+
+    pass
+
+
+class InventoryValidationError(InventoryError):
+    """inventory validation error"""
+
+    pass
+
+
+class InvalidTargetError(InventoryError):
+    """inventory validation error"""
+
+    pass