Inventory postprocessing filters

CHANGELOG.md

@@ -16,6 +16,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
   components with entries in `applications`.
   `classes: [ "components.argocd" ]` becomes `applications: [ "argocd" ]`.
 
+* Add option to define postprocessing filters in the Kapitan inventory ([#222]).
+
 ### Fixed
 
 * Replace remaining references to `common.yml` with `commodore.yml` ([#204])
@@ -248,6 +250,7 @@ Initial implementation
 [#212]: https://github.com/projectsyn/commodore/pull/212
 [#215]: https://github.com/projectsyn/commodore/pull/215
 [#217]: https://github.com/projectsyn/commodore/pull/217
+[#222]: https://github.com/projectsyn/commodore/pull/222
 [#226]: https://github.com/projectsyn/commodore/pull/226
 [#227]: https://github.com/projectsyn/commodore/pull/227
 [#237]: https://github.com/projectsyn/commodore/pull/237

commodore/postprocess/__init__.py

@@ -1,57 +1,182 @@
-from typing import Dict
+from pathlib import Path as P
+from typing import Any, Callable, ClassVar, Dict, List, Set
+from typing_extensions import Protocol
 
 import click
 
-from commodore.component import Component
+from commodore.config import Config, Component
 from commodore.helpers import yaml_load
 
-from .jsonnet import run_jsonnet_filter
-from .builtin_filters import run_builtin_filter
 from .inventory import resolve_inventory_vars, InventoryError
 
+from .jsonnet import run_jsonnet_filter, validate_jsonnet_filter
+from .builtin_filters import run_builtin_filter, validate_builtin_filter
 
-def postprocess_components(config, kapitan_inventory, components: Dict[str, Component]):
+
+class FilterFunc(Protocol):
+    def __call__(
+        self, inventory: Dict, component: str, filterid: str, path: P, **filterargs: str
+    ):
+        ...
+
+
+ValidateFunc = Callable[[str, Dict], Dict]
+
+
+class Filter:
+    type: str
+    filter: str
+    path: P
+    filterargs: Dict
+    enabled: bool
+
+    # PyLint complains about ClassVar not being subscriptable
+    # pylint: disable=unsubscriptable-object
+    _run_handlers: ClassVar[Dict[str, FilterFunc]] = {
+        "builtin": run_builtin_filter,
+        "jsonnet": run_jsonnet_filter,
+    }
+    # pylint: disable=unsubscriptable-object
+    _validate_handlers: ClassVar[Dict[str, ValidateFunc]] = {
+        "builtin": validate_builtin_filter,
+        "jsonnet": validate_jsonnet_filter,
+    }
+    # pylint: disable=unsubscriptable-object
+    _required_keys: ClassVar[Set[str]] = {"type", "path", "filter"}
+
+    def __init__(self, fd: Dict):
+        """
+        Assumes that `fd` has been validated with `_validate_filter`.
+        """
+        self.type = fd["type"]
+        self.filter = fd["filter"]
+        self.path = P(fd["path"])
+        self.enabled = fd.get("enabled", True)
+        self.filterargs = fd.get("filterargs", {})
+        self._runner: FilterFunc = self._run_handlers[self.type]
+
+    def run(self, inventory: Dict, component: str):
+        """
+        Run the filter.
+        """
+        if not self.enabled:
+            click.secho(
+                f" > Skipping disabled filter {self.filter} on path {self.path}"
+            )
+            return
+
+        self._runner(inventory, component, self.filter, self.path, **self.filterargs)
+
+    @classmethod
+    def validate(cls, cn: str, f: Dict):
+        """
+        Validate filter definition in `f`.
+        Raises exceptions as appropriate when the definition is invalid.
+        Returns the definiton if it validates successfully.
+        """
+        if not all(key in f for key in cls._required_keys):
+            missing_required_keys = cls._required_keys - f.keys()
+            raise KeyError(f"Filter is missing required key(s) {missing_required_keys}")
+
+        if "enabled" in f and not isinstance(f["enabled"], bool):
+            raise ValueError("Filter key 'enabled' is not a boolean")
+
+        typ = f["type"]
+        if typ not in cls._validate_handlers:
+            raise ValueError(f"Filter has unknown type {typ}")
+
+        # perform type-specific extra validation
+        cls._validate_handlers[typ](cn, f)
+
+        return f
+
+    @classmethod
+    def from_dict(cls, cn: str, f: Dict):
+        """
+        Create Filter object from filter definition dict `f`.
+        Raises exceptions as appropriate when the definition is invalid.
+        Returns a Filter object if the passed definition validates successfully.
+        """
+        return Filter(Filter.validate(cn, f))
+
+
+def _get_inventory_filters(inventory: Dict[str, Any]) -> List[Dict[str, Any]]:
+    """
+    Return list of filters defined in inventory.
+
+    Inventory filters are expected to be defined as a list in
+    `parameters.commodore.postprocess.filters`.
+    """
+    commodore = inventory["parameters"].get("commodore", {})
+    return commodore.get("postprocess", {}).get("filters", [])
+
+
+def _get_external_filters(
+    inventory: Dict[str, Any], c: Component
+) -> List[Dict[str, Any]]:
+    filters_file = c.filters_file
+    filters = []
+    if filters_file.is_file():
+        _filters = yaml_load(filters_file).get("filters", [])
+        for f in _filters:
+            # Resolve any inventory references in filter definition
+            try:
+                f = resolve_inventory_vars(inventory, f)
+            except InventoryError as e:
+                raise click.ClickException(
+                    f"Failed to resolve reclass references for external filter: {e}"
+                ) from e
+
+            # external filters without 'type' always have type 'jsonnet'
+            if "type" not in f:
+                click.secho(
+                    "   > [WARN] component uses untyped external postprocessing filter",
+                    fg="yellow",
+                )
+                f["type"] = "jsonnet"
+
+            if f["type"] == "jsonnet":
+                f["path"] = f["output_path"]
+                del f["output_path"]
+                f["filter"] = str(P("postprocess") / f["filter"])
+            filters.append(f)
+
+    return filters
+
+
+def postprocess_components(
+    config: Config,
+    kapitan_inventory: Dict[str, Dict[str, Any]],
+    components: Dict[str, Component],
+):
     click.secho("Postprocessing...", bold=True)
+
     for cn, c in components.items():
         inventory = kapitan_inventory.get(cn)
         if not inventory:
             click.echo(f" > No target exists for component {cn}, skipping...")
             continue
 
-        filters_file = c.filters_file
-        if filters_file.is_file():
+        # inventory filters
+        invfilters = _get_inventory_filters(inventory)
+
+        # "old", external filters
+        extfilters = _get_external_filters(inventory, c)
+
+        filters: List[Filter] = []
+        for fd in invfilters + extfilters:
+            try:
+                filters.append(Filter.from_dict(cn, fd))
+            except (KeyError, ValueError) as e:
+                click.secho(
+                    f" > Skipping filter '{fd['filter']}' with invalid definition {fd}: {e}",
+                    fg="yellow",
+                )
+
+        if len(filters) > 0 and config.debug:
+            click.echo(f" > {cn}...")
+
+        for f in filters:
             if config.debug:
-                click.echo(f" > {cn}...")
-            filters = yaml_load(filters_file)
-            for f in filters["filters"]:
-                # Resolve any inventory references in filter definition
-                try:
-                    f = resolve_inventory_vars(inventory, f)
-                except InventoryError as e:
-                    raise click.ClickException(
-                        f"Failed to resolve variables for old-style filter: {e}"
-                    ) from e
-
-                # filters without 'type' are always 'jsonnet'
-                if "type" not in f:
-                    click.secho(
-                        "   > [WARN] component uses old-style postprocess filters",
-                        fg="yellow",
-                    )
-                    f["type"] = "jsonnet"
-                # Filters which aren't explicitly disabled are always enabled
-                if "enabled" in f and not f["enabled"]:
-                    click.secho(
-                        "   > Skipping disabled filter"
-                        + f" {f['filter']} on path {f['path']}"
-                    )
-                    continue
-                if f["type"] == "jsonnet":
-                    run_jsonnet_filter(inventory, cn, filters_file.parent, f)
-                elif f["type"] == "builtin":
-                    run_builtin_filter(inventory, cn, f)
-                else:
-                    click.secho(
-                        f"   > [WARN] unknown builtin filter {f['filter']}",
-                        fg="yellow",
-                    )
+                click.secho(f"   > Executing filter '{f.type}:{f.filter}'")
+            f.run(inventory, cn)

commodore/postprocess/builtin_filters.py

@@ -1,6 +1,7 @@
 import json
 
 from pathlib import Path as P
+from typing import Dict
 
 import _jsonnet
 import click
@@ -10,15 +11,23 @@
 from .jsonnet import jsonnet_runner
 
 
-def _builtin_filter_helm_namespace(inv, component, path, **kwargs):
+def _output_dir(component: str, path):
+    return P("compiled", component, path)
+
+
+def _builtin_filter_helm_namespace(inv, component: str, path, **kwargs):
     if "namespace" not in kwargs:
         raise click.ClickException(
             "Builtin filter 'helm_namespace': filter argument 'namespace' is required"
         )
     create_namespace = kwargs.get("create_namespace", "false")
+    # Transform create_namespace to string as jsonnet extvars can only be
+    # strings
+    if isinstance(create_namespace, bool):
+        create_namespace = "true" if create_namespace else "false"
     exclude_objects = kwargs.get("exclude_objects", [])
     exclude_objects = "|".join([json.dumps(e) for e in exclude_objects])
-    output_dir = P("compiled", component, path)
+    output_dir = _output_dir(component, path)
 
     # pylint: disable=c-extension-no-member
     jsonnet_runner(
@@ -39,9 +48,27 @@ def _builtin_filter_helm_namespace(inv, component, path, **kwargs):
 }
 
 
-def run_builtin_filter(inv, component, f):
-    fname = f["filter"]
-    if fname not in _builtin_filters:
-        click.secho(f"   > [ERR ] Unknown builtin filter {fname}", fg="red")
-        return
-    _builtin_filters[fname](inv, component, f["path"], **f["filterargs"])
+class UnknownBuiltinFilter(ValueError):
+    def __init__(self, filtername):
+        super().__init__(f"Unknown builtin filter: {filtername}")
+        self.filtername = filtername
+
+
+def run_builtin_filter(
+    inventory: Dict, component: str, filterid: str, path: P, **filterargs: str
+):
+    if filterid not in _builtin_filters:
+        raise UnknownBuiltinFilter(filterid)
+    _builtin_filters[filterid](inventory, component, path, **filterargs)
+
+
+def validate_builtin_filter(cn: str, fd: Dict):
+    if fd["filter"] not in _builtin_filters:
+        raise UnknownBuiltinFilter(fd["filter"])
+
+    if "filterargs" not in fd:
+        raise KeyError("Builtin filter is missing required key 'filterargs'")
+
+    fpath = _output_dir(cn, fd["path"])
+    if not fpath.exists():
+        raise ValueError("Builtin filter called on path which doesn't exist")