Skip to content

Commit

Permalink
Add commands for evaluating HelmReleases (#12)
Browse files Browse the repository at this point in the history
Add commands for evaluating HelmReleases. The command can run `kustomize
build` to locate `HelmRelease` and `HelmRepository` objects, then
evaluate them with `helm template`. The resulting objects can also be
filtered using `kustomize cfg grep`.
allenporter authored Feb 6, 2023
1 parent f3b8663 commit 3da0d7d
Showing 16 changed files with 411 additions and 109 deletions.
19 changes: 9 additions & 10 deletions .pre-commit-config.yaml
Original file line number Diff line number Diff line change
@@ -19,28 +19,27 @@ repos:
- -c
- ".yaml-lint.yaml"
- repo: https://github.com/pycqa/isort
rev: 5.9.3
rev: 5.12.0
hooks:
- id: isort
args: ["--profile", "black"]
- repo: https://github.com/psf/black
rev: 22.3.0
hooks:
- id: black
- repo: https://gitlab.com/pycqa/flake8
rev: 4.0.1
- repo: https://github.com/PyCQA/flake8
rev: 6.0.0
hooks:
- id: flake8
- repo: https://github.com/pre-commit/mirrors-mypy
rev: v0.991
- repo: local
hooks:
- id: mypy
name: mypy
entry: script/run-in-env.sh mypy
language: script
types: [python]
require_serial: true
files: ^(flux_local/|tests/)
additional_dependencies:
- "pydantic>=1.10.4"
- "types-PyYAML>=6.0.12.4"
- repo: local
hooks:
- id: pylint
name: pylint
entry: pylint
135 changes: 129 additions & 6 deletions flux_local/helm.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,132 @@
"""Library for running `helm template` to produce local items in the cluster."""
"""Library for running `helm template` to produce local items in the cluster.
import subprocess
You can instantiate a helm template with the following:
- A HelmRepository which is a url that contains charts
- A HelmRelease which is an instance of a HelmChart in a HelmRepository
This is an example that prepares the helm repository:
```
from flux_local.kustomize import Kustomize
from flux_local.helm import Helm
from flux_local.manifest import HelmRepository
def render(chart, values):
"""Runs `helm template` to produce local items in the cluster."""
cmd = ["helm", "template", chart, "--values", values]
return subprocess.check_output(cmd)
kustomize = Kustomize.build(TESTDATA_DIR)
repos = await kustomize.grep("kind=^HelmRepository$").objects()
helm = Helm("/tmp/path/helm", "/tmp/path/cache")
for repo in repos:
helm.add_repo(HelmRepository.from_doc(repo))
await helm.update()
```
Then to actually instantiate a template from a HelmRelease:
```
from flux_local.manifest import HelmRelease
releases = await kustomize.grep("kind=^HelmRelease$").objects()
if not len(releases) == 1:
raise ValueError("Expected only one HelmRelease")
tmpl = helm.template(
HelmRelease.from_doc(releases[0]),
releases[0]["spec"].get("values"))
objects = await tmpl.objects()
for object in objects:
print(f"Found object {object['apiVersion']} {object['kind']}")
```
"""

import datetime
import logging
from pathlib import Path
from typing import Any

import aiofiles
import yaml

from . import command
from .kustomize import Kustomize
from .manifest import HelmRelease, HelmRepository

__all__ = [
"Helm",
]

_LOGGER = logging.getLogger(__name__)


HELM_BIN = "helm"


class RepositoryConfig:
"""Generates a helm repository configuration from flux HelmRepository objects."""

def __init__(self, repos: list[HelmRepository]) -> None:
"""Initialize RepositoryConfig."""
self._repos = repos

@property
def config(self) -> dict[str, Any]:
"""Return a synthetic repository config object."""
now = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0)
return {
"apiVersion": "",
"generated": now.isoformat(),
"repositories": [
{
"name": f"{repo.namespace}-{repo.name}",
"url": repo.url,
}
for repo in self._repos
],
}


class Helm:
"""Manages local HelmRepository state."""

def __init__(self, tmp_dir: Path, cache_dir: Path) -> None:
"""Initialize Helm."""
self._tmp_dir = tmp_dir
self._repo_config_file = self._tmp_dir / "repository-config.yaml"
self._flags = [
"--registry-config",
"/dev/null",
"--repository-cache",
str(cache_dir),
"--repository-config",
str(self._repo_config_file),
]
self._repos: list[HelmRepository] = []

def add_repo(self, repo: HelmRepository) -> None:
"""Add the specified HelmRepository to the local config."""
self._repos.append(repo)

async def update(self) -> None:
"""Return command line arguments to update the local repo.
Typically the repository must be updated before doing any chart templating.
"""
content = yaml.dump(RepositoryConfig(self._repos).config)
async with aiofiles.open(str(self._repo_config_file), mode="w") as config_file:
await config_file.write(content)
await command.run([HELM_BIN, "repo", "update"] + self._flags)

async def template(self, release: HelmRelease, values: dict[str, Any]) -> Kustomize:
"""Return command line arguments to template the specified chart."""
args = [
HELM_BIN,
"template",
release.name,
release.chart.chart_name,
"--namespace",
release.namespace,
"--skip-crds", # Reduce size of output
"--version",
release.chart.version,
]
if values:
values_path = self._tmp_dir / f"{release.release_name}-values.yaml"
async with aiofiles.open(values_path, mode="w") as values_file:
await values_file.write(yaml.dump(values))
args.extend(["--values", str(values_path)])
return Kustomize([args + self._flags])
77 changes: 61 additions & 16 deletions flux_local/kustomize.py
Original file line number Diff line number Diff line change
@@ -3,30 +3,75 @@
Kustomize build can be used to apply overlays and output a set of resources or artifacts
for the cluster that can be either be parsed directly or piped into additional
commands for processing and filtering by kustomize grep.
This example returns the objects inside a Kustomization using `kustomize build`:
```
from flux_local.kustomize import Kustomize
objects = await Kustomize.build('/path/to/objects').objects()
for object in objects:
print(f"Found object {object['apiVersion']} {object['kind']}")
```
You can also filter documents to specific resource types or other fields:
```
from flux_local.kustomize import Kustomize
objects = await Kustomize.build('/path/to/objects').grep('kind=ConfigMap').objects()
for object in objects:
print(f"Found ConfigMap: {object['metadata']['name']}")
```
"""

from pathlib import Path
from typing import Any, AsyncGenerator

import yaml

from . import command

KUSTOMIZE_BIN = "kustomize"


def build(path: Path) -> list[str]:
"""Generate a kustomize build command for the specified path."""
return [KUSTOMIZE_BIN, "build", str(path)]
class Kustomize:
"""Library for issuing a kustomize command."""

def __init__(self, cmds: list[list[str]]) -> None:
"""Initialize Kustomize."""
self._cmds = cmds

@classmethod
def build(cls, path: Path) -> "Kustomize":
"""Build cluster artifacts from the specified path."""
return Kustomize(cmds=[[KUSTOMIZE_BIN, "build", str(path)]])

def grep(
self, expr: str, path: Path | None = None, invert: bool = False
) -> "Kustomize":
"""Filter resources based on an expression.
Example expressions:
`kind=HelmRelease`
`metadata.name=redis`
"""
out = [KUSTOMIZE_BIN, "cfg", "grep", expr]
if invert:
out.append("--invert-match")
if path:
out.append(str(path))
return Kustomize(self._cmds + [out])

def grep(expr: str, path: Path | None = None, invert: bool = False) -> list[str]:
"""Generate a kustomize grep command to filter resources based on an expression.
async def run(self) -> str:
"""Run the kustomize command and return the output as a string."""
return await command.run_piped(self._cmds)

Example expressions:
`kind=HelmRelease`
`metadata.name=redis`
async def _docs(self) -> AsyncGenerator[dict[str, Any], None]:
"""Run the kustomize command and return the result documents."""
out = await self.run()
for doc in yaml.safe_load_all(out):
yield doc

The return value is a set of command args.
"""
out = [KUSTOMIZE_BIN, "cfg", "grep", expr]
if invert:
out.append("--invert-match")
if path:
out.append(str(path))
return out
async def objects(self) -> list[dict[str, Any]]:
"""Run the kustomize command and return the result cluster objects as a list."""
return [doc async for doc in self._docs()]
66 changes: 48 additions & 18 deletions flux_local/manifest.py
Original file line number Diff line number Diff line change
@@ -17,18 +17,19 @@
"Kustomization",
"HelmRepository",
"HelmRelease",
"HelmChart",
]


@dataclass
class HelmRelease:
"""A representation of a Flux HelmRelease."""
class HelmChart:
"""A representation of an instantiation of a chart for a HelmRelease."""

name: str
"""The name of the HelmRelease."""
"""The name of the chart within the HelmRepository."""

namespace: str
"""The namespace that owns the HelmRelease."""
version: str
"""The version of the chart."""

repo_name: str
"""The name of the HelmRepository."""
@@ -37,29 +38,58 @@ class HelmRelease:
"""The namespace of the HelmRepository."""

@classmethod
def from_doc(cls, doc: dict[str, Any]) -> "HelmRelease":
"""Parse a HelmRelease from a kubernetes resource object."""
if not (metadata := doc.get("metadata")):
raise ValueError(f"Invalid {cls} missing metadata: {doc}")
if not (name := metadata.get("name")):
raise ValueError(f"Invalid {cls} missing metadata.name: {doc}")
if not (namespace := metadata.get("namespace")):
raise ValueError(f"Invalid {cls} missing metadata.namespace: {doc}")
def from_doc(cls, doc: dict[str, Any]) -> "HelmChart":
"""Parse a HelmChart from a HelmRelease resource object."""
if not (spec := doc.get("spec")):
raise ValueError(f"Invalid {cls} missing spec: {doc}")
if not (chart := spec.get("chart")):
raise ValueError(f"Invalid {cls} missing spec.chart: {doc}")
if not (chart_spec := chart.get("spec")):
raise ValueError(f"Invalid {cls} missing spec.chart.spec: {doc}")
if not (chart := chart_spec.get("chart")):
raise ValueError(f"Invalid {cls} missing spec.chart.spec.chart: {doc}")
if not (version := chart_spec.get("version")):
raise ValueError(f"Invalid {cls} missing spec.chart.spec.version: {doc}")
if not (source_ref := chart_spec.get("sourceRef")):
raise ValueError(f"Invalid {cls} missing spec.chart.spec.sourceRef: {doc}")
if "namespace" not in source_ref or "name" not in source_ref:
raise ValueError(f"Invalid {cls} missing sourceRef fields: {doc}")
return cls(name, namespace, source_ref["name"], source_ref["namespace"])
return cls(chart, version, source_ref["name"], source_ref["namespace"])

@property
def id_name(self) -> str:
"""Identifier for the HelmRelease in tests."""
def chart_name(self) -> str:
"""Identifier for the HelmChart."""
return f"{self.repo_namespace}-{self.repo_name}/{self.name}"


@dataclass
class HelmRelease:
"""A representation of a Flux HelmRelease."""

name: str
"""The name of the HelmRelease."""

namespace: str
"""The namespace that owns the HelmRelease."""

chart: HelmChart
"""A mapping to a specific helm chart for this HelmRelease."""

@classmethod
def from_doc(cls, doc: dict[str, Any]) -> "HelmRelease":
"""Parse a HelmRelease from a kubernetes resource object."""
if not (metadata := doc.get("metadata")):
raise ValueError(f"Invalid {cls} missing metadata: {doc}")
if not (name := metadata.get("name")):
raise ValueError(f"Invalid {cls} missing metadata.name: {doc}")
if not (namespace := metadata.get("namespace")):
raise ValueError(f"Invalid {cls} missing metadata.namespace: {doc}")
chart = HelmChart.from_doc(doc)
return cls(name, namespace, chart)

@property
def release_name(self) -> str:
"""Identifier for the HelmRelease."""
return f"{self.namespace}-{self.name}"


@@ -92,8 +122,8 @@ def from_doc(cls, doc: dict[str, Any]) -> "HelmRepository":
return cls(name, namespace, url)

@property
def id_name(self) -> str:
"""Identifier for the HelmRepository in tests."""
def repo_name(self) -> str:
"""Identifier for the HelmRepository."""
return f"{self.namespace}-{self.name}"


1 change: 1 addition & 0 deletions requirements.txt
Original file line number Diff line number Diff line change
@@ -16,6 +16,7 @@ pytest-asyncio==0.20.3
pytest-cov==3.0.0
python-slugify==8.0.0
PyYAML==6.0
types-aiofiles==22.1.0.6
types-PyYAML==6.0.12.4
typing-extensions==4.4.0
wheel==0.37.1
Loading

0 comments on commit 3da0d7d

Please sign in to comment.