flux_local.helm

  1"""Library for running `helm template` to produce local items in the cluster.
  2
  3You can instantiate a helm template with the following:
  4- A HelmRepository which is a url that contains charts
  5- A HelmRelease which is an instance of a HelmChart in a HelmRepository
  6
  7This is an example that prepares the helm repository:
  8```python
  9from flux_local.kustomize import Kustomize
 10from flux_local.helm import Helm
 11from flux_local.manifest import HelmRepository
 12
 13kustomize = Kustomize.build(TESTDATA_DIR)
 14repos = await kustomize.grep("kind=^HelmRepository$").objects()
 15helm = Helm("/tmp/path/helm", "/tmp/path/cache")
 16for repo in repos:
 17  helm.add_repo(HelmRepository.parse_doc(repo))
 18await helm.update()
 19```
 20
 21Then to actually instantiate a template from a HelmRelease:
 22```python
 23from flux_local.manifest import HelmRelease
 24
 25releases = await kustomize.grep("kind=^HelmRelease$").objects()
 26if not len(releases) == 1:
 27    raise ValueError("Expected only one HelmRelease")
 28tmpl = helm.template(HelmRelease.parse_doc(releases[0]))
 29objects = await tmpl.objects()
 30for object in objects:
 31    print(f"Found object {object['apiVersion']} {object['kind']}")
 32```
 33"""
 34
 35from collections.abc import Generator
 36from contextlib import contextmanager
 37import contextvars
 38import datetime
 39from dataclasses import dataclass
 40import logging
 41from pathlib import Path
 42import tempfile
 43from typing import Any
 44
 45import aiofiles
 46from aiofiles.ospath import exists
 47import yaml
 48
 49from . import command
 50from .kustomize import Kustomize
 51from .manifest import (
 52    HelmRelease,
 53    HelmRepository,
 54    OCIRepository,
 55    CRD_KIND,
 56    SECRET_KIND,
 57    REPO_TYPE_OCI,
 58    HELM_REPO_KIND,
 59    GIT_REPOSITORY,
 60    OCI_REPOSITORY,
 61    GitRepository,
 62)
 63from .source_controller.artifact import GitArtifact
 64from .exceptions import HelmException
 65
 66__all__ = [
 67    "Helm",
 68    "Options",
 69]
 70
 71_LOGGER = logging.getLogger(__name__)
 72
 73
 74HELM_BIN = "helm"
 75
 76_config_context = contextvars.ContextVar[str | None]("config_context", default=None)
 77
 78
 79@dataclass(kw_only=True, frozen=True)
 80class LocalGitRepository:
 81    """A GitRepository resolved to a local path.."""
 82
 83    repo: GitRepository
 84    """The GitRepository object."""
 85
 86    artifact: GitArtifact
 87    """The GitArtifact object."""
 88
 89    @property
 90    def repo_name(self) -> str:
 91        """Return the name of the repository."""
 92        return self.repo.repo_name
 93
 94
 95@contextmanager
 96def empty_registry_config_file() -> Generator[Path]:
 97    """Context manager a temporary JSON configuration file.
 98
 99    This is needed because some versions of helm can't handle reading /dev/null.
100    It is preferred to call this once at the start of the program to create the
101    empty json file. It may be called multiple times and it will reuse the
102    existing file.
103    """
104    if (existing_path := _config_context.get()) is not None:
105        # Reuse existing config file already created
106        yield Path(existing_path)
107        return
108
109    with tempfile.NamedTemporaryFile(
110        mode="w+",
111        suffix=".json",
112    ) as temp_file:
113        temp_file_path = Path(temp_file.name)
114        temp_file_path.write_text("{}")
115        token = _config_context.set(str(temp_file_path))
116        try:
117            yield temp_file_path
118        finally:
119            _config_context.reset(token)
120
121
122def _get_registry_config_file() -> str:
123    """Get the current registry config file."""
124    if (filename := _config_context.get()) is None:
125        raise ValueError(
126            "No registry config file found, call with empty_registry_config_file() first"
127        )
128    return filename
129
130
131def _chart_name(
132    release: HelmRelease,
133    repo: HelmRepository | OCIRepository | LocalGitRepository | None,
134) -> str:
135    """Return the helm chart name used for the helm template command."""
136    if release.chart.repo_kind == OCI_REPOSITORY:
137        if not repo:
138            raise HelmException(
139                f"Unable to find OCIRepository for {release.chart.chart_name} for "
140                f"HelmRelease {release.name}"
141            )
142        if isinstance(repo, OCIRepository):
143            return repo.url
144        raise HelmException(
145            f"HelmRelease {release.name} expected OCIRepository but got HelmRepository {repo.repo_name}"
146        )
147    if release.chart.repo_kind == HELM_REPO_KIND:
148        if not repo:
149            raise HelmException(
150                f"Unable to find HelmRepository for {release.chart.chart_name} for "
151                f"HelmRelease {release.name}"
152            )
153        if not isinstance(repo, HelmRepository):
154            raise HelmException(
155                f"HelmRelease {release.name} expected HelmRepository but got OCIRepository {repo.repo_name}"
156            )
157        return repo.helm_chart_name(release.chart)
158    if release.chart.repo_kind == GIT_REPOSITORY:
159        if isinstance(repo, LocalGitRepository):
160            return str(Path(repo.artifact.local_path) / release.chart.name)
161        _LOGGER.warning(
162            "Unable to find chart %s for HelmRelease %s, using %s",
163            release.chart.chart_name,
164            release.name,
165            release.chart.name,
166        )
167        return release.chart.name
168    raise HelmException(
169        f"Unable to find chart source for chart {release.chart.chart_name} "
170        f"kind {release.chart.repo_kind} for HelmRelease {release.name}"
171    )
172
173
174class RepositoryConfig:
175    """Generates a helm repository configuration from flux HelmRepository objects."""
176
177    def __init__(self, repos: list[HelmRepository]) -> None:
178        """Initialize RepositoryConfig."""
179        self._repos = repos
180
181    @property
182    def config(self) -> dict[str, Any]:
183        """Return a synthetic repository config object."""
184        now = datetime.datetime.now(datetime.timezone.utc).replace(microsecond=0)
185        return {
186            "apiVersion": "",
187            "generated": now.isoformat(),
188            "repositories": [
189                {
190                    "name": f"{repo.namespace}-{repo.name}",
191                    "url": repo.url,
192                }
193                for repo in self._repos
194            ],
195        }
196
197
198@dataclass
199class Options:
200    """Options to use when inflating a Helm chart.
201
202    Internally, these translate into either command line flags or
203    resource kinds that will be filtered form the output.
204    """
205
206    skip_crds: bool = True
207    """Skip CRDs when building the output."""
208
209    skip_tests: bool = True
210    """Don't run helm tests on the output."""
211
212    skip_secrets: bool = False
213    """Don't emit secrets in the output."""
214
215    skip_kinds: list[str] | None = None
216    """Omit these kinds in the output."""
217
218    kube_version: str | None = None
219    """Value of the helm --kube-version flag."""
220
221    api_versions: str | None = None
222    """Value of the helm --api-versions flag."""
223
224    registry_config: str | None = None
225    """Value of the helm --registry-config flag."""
226
227    skip_invalid_paths: bool = False
228    """Skip HelmReleases with invalid local paths."""
229
230    @property
231    def base_args(self) -> list[str]:
232        """Helm template CLI arguments built from the options."""
233        return [
234            "--registry-config",
235            self.registry_config or _get_registry_config_file(),
236        ]
237
238    @property
239    def template_args(self) -> list[str]:
240        """Helm template CLI arguments built from the options."""
241        args = self.base_args
242        if self.skip_crds:
243            args.append("--skip-crds")
244        if self.skip_tests:
245            args.append("--skip-tests")
246        if self.kube_version:
247            args.extend(["--kube-version", self.kube_version])
248        if self.api_versions:
249            args.extend(["--api-versions", self.api_versions])
250        return args
251
252    @property
253    def skip_resources(self) -> list[str]:
254        """A list of CRD resources to filter from the output."""
255        skips = []
256        if self.skip_crds:
257            skips.append(CRD_KIND)
258        if self.skip_secrets:
259            skips.append(SECRET_KIND)
260        if self.skip_kinds:
261            skips.extend(self.skip_kinds)
262        return skips
263
264
265class Helm:
266    """Manages local HelmRepository state."""
267
268    def __init__(self, tmp_dir: Path, cache_dir: Path) -> None:
269        """Initialize Helm."""
270        self._tmp_dir = tmp_dir
271        self._repo_config_file = self._tmp_dir / "repository-config.yaml"
272        self._flags = [
273            "--repository-cache",
274            str(cache_dir),
275            "--repository-config",
276            str(self._repo_config_file),
277        ]
278        self._repos: list[HelmRepository | OCIRepository | LocalGitRepository] = []
279
280    def add_repo(
281        self, repo: HelmRepository | OCIRepository | LocalGitRepository
282    ) -> None:
283        """Add the specified HelmRepository to the local config."""
284        self._repos.append(repo)
285
286    def add_repos(
287        self,
288        repos: (
289            list[HelmRepository]
290            | list[OCIRepository]
291            | list[HelmRepository | OCIRepository]
292            | list[HelmRepository | OCIRepository | LocalGitRepository]
293        ),
294    ) -> None:
295        """Add the specified HelmRepository to the local config."""
296        for repo in repos:
297            self._repos.append(repo)
298
299    async def update(self) -> None:
300        """Run the update command to update the local repo.
301
302        Typically the repository must be updated before doing any chart templating.
303        """
304        _LOGGER.debug("Updating %d repositories", len(self._repos))
305        helm_repos = [
306            repo
307            for repo in self._repos
308            if isinstance(repo, HelmRepository) and repo.repo_type != REPO_TYPE_OCI
309        ]
310        if not helm_repos:
311            return
312        content = yaml.dump(RepositoryConfig(helm_repos).config, sort_keys=False)
313        async with aiofiles.open(str(self._repo_config_file), mode="w") as config_file:
314            await config_file.write(content)
315        with empty_registry_config_file():
316            args = [HELM_BIN, "repo", "update"]
317            args.extend(Options().base_args)
318            args.extend(self._flags)
319            await command.run(command.Command(args, exc=HelmException))
320
321    async def is_invalid_local_path(
322        self,
323        release: HelmRelease,
324    ) -> bool:
325        """Check if the HelmRelease has an invalid GitRepository path."""
326        repo = next(
327            iter([repo for repo in self._repos if repo.repo_name == release.repo_name]),
328            None,
329        )
330        chart_name = _chart_name(release, repo)
331        if release.chart.repo_kind == GIT_REPOSITORY:
332            if not await exists(chart_name):
333                return True
334        return False
335
336    async def template(
337        self,
338        release: HelmRelease,
339        options: Options | None = None,
340    ) -> Kustomize:
341        """Return command to evaluate the template of the specified chart.
342
343        The values will come from the `HelmRelease` object.
344        """
345        if options is None:
346            options = Options()
347        repo = next(
348            iter([repo for repo in self._repos if repo.repo_name == release.repo_name]),
349            None,
350        )
351        with empty_registry_config_file():
352            args: list[str] = [
353                HELM_BIN,
354                "template",
355                release.name,
356                _chart_name(release, repo),
357                "--namespace",
358                release.release_namespace,
359            ]
360            args.extend(self._flags)
361            args.extend(options.template_args)
362            if release.disable_openapi_validation:
363                args.append("--disable-openapi-validation")
364            if release.disable_schema_validation:
365                args.append("--skip-schema-validation")
366            if release.chart.version:
367                args.extend(
368                    [
369                        "--version",
370                        release.chart.version,
371                    ]
372                )
373            elif isinstance(repo, OCIRepository) and (oci_version := repo.version()):
374                args.extend(
375                    [
376                        "--version",
377                        oci_version,
378                    ]
379                )
380            if release.values:
381                values_path = self._tmp_dir / f"{release.release_name}-values.yaml"
382                async with aiofiles.open(values_path, mode="w") as values_file:
383                    await values_file.write(yaml.dump(release.values, sort_keys=False))
384                args.extend(["--values", str(values_path)])
385            cmd = Kustomize([command.Command(args, exc=HelmException)])
386            if options.skip_resources:
387                cmd = cmd.skip_resources(options.skip_resources)
388            return cmd