mirror of
https://github.com/kevinveenbirkenbach/computer-playbook.git
synced 2026-11-06 10:59:59 +00:00
kevinveenbirkenbachandClaude Opus 5
a650801b35
compose_volumes copied extra_configs and extra_secrets through verbatim.
A file-backed extra without a name therefore reached swarm under a fixed
name, and swarm secrets are immutable: a new content under the same name
is rejected on the next stack deploy. Such extras now get the same
{role}_{key}_{sha8(content)} name that _config_secret_name already gives
meta-declared entries. An explicit name is kept.
container_volumes appended extras as handed over. Ansible passes one
lookup's result to another as proxy containers, and a TaggedInt mode
inside a proxy mapping made PyYAML raise "cannot represent an object".
The private _to_plain of compose_volumes moves to utils/templating as
to_plain, now coercing bool, int and float to the exact builtin types
(bool before int, since bool subclasses int), and container_volumes runs
its extras through it.
Only web-app-matrix passes extra secrets or configs today; the
extra_volumes path that web-app-nextcloud uses is unchanged.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
274 lines
10 KiB
Python
274 lines
10 KiB
Python
"""Lookup ``container_volumes(service)``: render the per-service
|
|
``volumes:`` / ``configs:`` / ``secrets:`` block fed from
|
|
``meta/volumes.yml``; ``application_id`` comes from the templating
|
|
context. Output starts at column 0; callers add the indent
|
|
(``| indent(4)`` or whatever depth) so the block sits under the right
|
|
service in ``compose.yml.j2``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING, Any
|
|
|
|
from ansible.errors import AnsibleError, AnsibleFilterError, AnsibleUndefinedVariable
|
|
from ansible.plugins.loader import lookup_loader
|
|
from ansible.plugins.lookup import LookupBase
|
|
|
|
# nocheck: lookup-cache-import (raw-volume accessor: canonical meta/volumes.yml shape)
|
|
from utils.cache.applications import get_canonical_volumes
|
|
from utils.cache.yaml import dump_yaml_str
|
|
from utils.roles.applications.mounts import (
|
|
mount_default_read_only,
|
|
mount_when_passes,
|
|
normalize_volumes_meta,
|
|
)
|
|
from utils.templating.ansible import _trust_as_template, to_plain
|
|
|
|
if TYPE_CHECKING:
|
|
from collections.abc import Callable
|
|
|
|
|
|
def _coerce_mode_int(value: Any) -> int:
|
|
if isinstance(value, bool):
|
|
raise AnsibleFilterError(
|
|
f"container_volumes: mode must be an octal int or string, got bool: {value!r}"
|
|
)
|
|
if isinstance(value, int):
|
|
return value
|
|
text = str(value).strip()
|
|
if not text:
|
|
raise AnsibleFilterError("container_volumes: mode is empty")
|
|
try:
|
|
return int(text, 8)
|
|
except ValueError as exc:
|
|
raise AnsibleFilterError(
|
|
f"container_volumes: mode {value!r} is not a valid octal: {exc}"
|
|
) from exc
|
|
|
|
|
|
_SIZE_UNITS = {"": 1, "b": 1, "k": 1024, "m": 1024**2, "g": 1024**3}
|
|
|
|
|
|
def _coerce_size_bytes(value: Any) -> int:
|
|
"""Return a tmpfs size as the byte count the long-form mount demands.
|
|
|
|
A declaration reads ``512m`` because that is how a human writes it and how
|
|
the short form accepts it. The long form takes an integer, and compose
|
|
rejects the string outright rather than parsing it, so the suffix has to be
|
|
resolved here.
|
|
"""
|
|
if isinstance(value, bool):
|
|
raise AnsibleFilterError(
|
|
f"container_volumes: size must be an int or a size string, got bool: {value!r}"
|
|
)
|
|
if isinstance(value, int):
|
|
return value
|
|
text = str(value).strip().lower()
|
|
if not text:
|
|
raise AnsibleFilterError("container_volumes: size is empty")
|
|
number, unit = text[:-1], text[-1]
|
|
if unit.isdigit():
|
|
number, unit = text, ""
|
|
try:
|
|
return int(number) * _SIZE_UNITS[unit]
|
|
except (ValueError, KeyError) as exc:
|
|
raise AnsibleFilterError(
|
|
f"container_volumes: size {value!r} is not a byte count: {exc}"
|
|
) from exc
|
|
|
|
|
|
def _short_form_volume(source_name: str, target: str, read_only: bool) -> str:
|
|
suffix = ":ro" if read_only else ""
|
|
return f"{source_name}:{target}{suffix}"
|
|
|
|
|
|
def container_volumes(
|
|
applications: dict[str, Any],
|
|
application_id: str,
|
|
service: str,
|
|
*,
|
|
extra_volumes: list[Any] | None = None,
|
|
extra_configs: list[dict[str, Any]] | None = None,
|
|
extra_secrets: list[dict[str, Any]] | None = None,
|
|
render_jinja: Callable[[str], str] | None = None,
|
|
) -> str:
|
|
"""Return the per-service mount block (``volumes:`` + ``configs:`` +
|
|
``secrets:``) as a YAML string starting at column 0.
|
|
"""
|
|
if not isinstance(applications, dict):
|
|
raise AnsibleFilterError("container_volumes: 'applications' must be a dict")
|
|
if not isinstance(application_id, str) or not application_id.strip():
|
|
raise AnsibleFilterError(
|
|
"container_volumes: 'application_id' must be a non-empty string"
|
|
)
|
|
if not isinstance(service, str) or not service.strip():
|
|
raise AnsibleFilterError(
|
|
"container_volumes: 'service' must be a non-empty string"
|
|
)
|
|
|
|
role_data = applications.get(application_id) or {}
|
|
raw_meta_volumes = (
|
|
get_canonical_volumes(application_id) or role_data.get("volumes") or {}
|
|
)
|
|
canonical_entries = normalize_volumes_meta(raw_meta_volumes)
|
|
|
|
volumes_block: list[Any] = []
|
|
configs_block: list[dict[str, Any]] = []
|
|
secrets_block: list[dict[str, Any]] = []
|
|
|
|
for semantic_name, entry in canonical_entries.items():
|
|
vtype = entry.get("type", "volume")
|
|
volume_level_ro = bool(entry.get("read_only", mount_default_read_only(entry)))
|
|
|
|
for mount in entry.get("mounts") or []:
|
|
if not isinstance(mount, dict):
|
|
continue
|
|
if mount.get("service") != service:
|
|
continue
|
|
if not mount_when_passes(mount, render_jinja):
|
|
continue
|
|
|
|
target = mount.get("target")
|
|
if not target:
|
|
continue
|
|
target_text = str(target)
|
|
if render_jinja is not None and (
|
|
"{{" in target_text or "{%" in target_text
|
|
):
|
|
try:
|
|
target = str(render_jinja(target_text))
|
|
except AnsibleUndefinedVariable as exc:
|
|
raise AnsibleFilterError(
|
|
f"container_volumes: failed to render mount target "
|
|
f"{target_text!r} for application {application_id!r}, "
|
|
f"service {service!r}: {exc}"
|
|
) from exc
|
|
|
|
if vtype == "volume":
|
|
read_only = bool(mount.get("read_only", volume_level_ro))
|
|
volumes_block.append(
|
|
_short_form_volume(semantic_name, target, read_only)
|
|
)
|
|
|
|
elif vtype == "bind":
|
|
source = str(entry.get("source", ""))
|
|
if render_jinja is not None and ("{{" in source or "{%" in source):
|
|
try:
|
|
source = str(render_jinja(source))
|
|
except AnsibleUndefinedVariable as exc:
|
|
raise AnsibleFilterError(
|
|
f"container_volumes: failed to render bind source "
|
|
f"{source!r} for application {application_id!r}, "
|
|
f"service {service!r}: {exc}"
|
|
) from exc
|
|
read_only = bool(mount.get("read_only", volume_level_ro))
|
|
volumes_block.append(_short_form_volume(source, target, read_only))
|
|
|
|
elif vtype == "tmpfs":
|
|
tmpfs_entry: dict[str, Any] = {
|
|
"type": "tmpfs",
|
|
"target": target,
|
|
}
|
|
tmpfs_options: dict[str, Any] = {}
|
|
if mount.get("size") is not None:
|
|
tmpfs_options["size"] = _coerce_size_bytes(mount["size"])
|
|
if mount.get("mode") is not None:
|
|
tmpfs_options["mode"] = _coerce_mode_int(mount["mode"])
|
|
if tmpfs_options:
|
|
tmpfs_entry["tmpfs"] = tmpfs_options
|
|
volumes_block.append(tmpfs_entry)
|
|
|
|
elif vtype == "config":
|
|
cfg_ref: dict[str, Any] = {"source": semantic_name, "target": target}
|
|
mode_value = entry.get("mode")
|
|
if mode_value:
|
|
cfg_ref["mode"] = _coerce_mode_int(mode_value)
|
|
configs_block.append(cfg_ref)
|
|
|
|
elif vtype == "secret":
|
|
sec_ref: dict[str, Any] = {"source": semantic_name, "target": target}
|
|
mode_value = entry.get("mode")
|
|
if mode_value:
|
|
sec_ref["mode"] = _coerce_mode_int(mode_value)
|
|
secrets_block.append(sec_ref)
|
|
|
|
if extra_volumes:
|
|
volumes_block.extend(to_plain(extra_volumes))
|
|
if extra_configs:
|
|
configs_block.extend(to_plain(extra_configs))
|
|
if extra_secrets:
|
|
secrets_block.extend(to_plain(extra_secrets))
|
|
|
|
payload: dict[str, Any] = {}
|
|
if volumes_block:
|
|
payload["volumes"] = volumes_block
|
|
if configs_block:
|
|
payload["configs"] = configs_block
|
|
if secrets_block:
|
|
payload["secrets"] = secrets_block
|
|
|
|
if not payload:
|
|
return ""
|
|
|
|
return dump_yaml_str(payload).rstrip()
|
|
|
|
|
|
class LookupModule(LookupBase):
|
|
def run(
|
|
self,
|
|
terms: list[Any] | None,
|
|
variables: dict[str, Any] | None = None,
|
|
**kwargs: Any,
|
|
) -> list[str]:
|
|
if not terms or len(terms) != 1:
|
|
raise AnsibleError(
|
|
"container_volumes lookup requires exactly one term: the "
|
|
"service; application_id is read from the templating context"
|
|
)
|
|
service = str(terms[0]).strip()
|
|
if not service:
|
|
raise AnsibleError("container_volumes lookup: service must be non-empty")
|
|
|
|
vars_ = variables or getattr(self._templar, "available_variables", {}) or {}
|
|
templar = getattr(self, "_templar", None)
|
|
|
|
application_id = vars_.get("application_id")
|
|
if templar is not None and application_id is not None:
|
|
application_id = templar.template(application_id)
|
|
application_id = str(application_id or "").strip()
|
|
if not application_id:
|
|
raise AnsibleError(
|
|
"container_volumes lookup: 'application_id' is not set in the "
|
|
"templating context"
|
|
)
|
|
|
|
applications = lookup_loader.get(
|
|
"applications", loader=self._loader, templar=getattr(self, "_templar", None)
|
|
).run([], variables=vars_)[0]
|
|
|
|
render_jinja = None
|
|
if templar is not None:
|
|
# Closure captures the templar so mount-level `when:` Jinja
|
|
# expressions are evaluated against the same context the
|
|
# caller template sees, including `{% set %}` vars.
|
|
def render_jinja(expr: str) -> Any:
|
|
if not isinstance(expr, str):
|
|
return expr
|
|
return templar.template(
|
|
_trust_as_template(expr),
|
|
fail_on_undefined=False,
|
|
)
|
|
|
|
rendered = container_volumes(
|
|
applications,
|
|
application_id,
|
|
service,
|
|
extra_volumes=kwargs.get("extra_volumes"),
|
|
extra_configs=kwargs.get("extra_configs"),
|
|
extra_secrets=kwargs.get("extra_secrets"),
|
|
render_jinja=render_jinja,
|
|
)
|
|
if rendered and not rendered.startswith("\n"):
|
|
rendered = "\n" + rendered
|
|
return [rendered]
|