Files
computer-playbook/plugins/lookup/timeout.py
kevinveenbirkenbachandClaude Opus 4.8 96aa7784b6 feat(sys-svc-compose): scale all task timeouts via TIMEOUT_FACTOR lookup
Add a 'timeout' lookup plugin that returns round(base * TIMEOUT_FACTOR)
(group_vars default 1, so value-identical to the old literals). One knob
in group_vars stretches every wait in the stack for a slow uplink:
TIMEOUT_FACTOR: 2 doubles them all.

- plugins/lookup/timeout.py + unit tests (identity, factor scaling,
  string coercion, error paths).
- group_vars/all/17_resource.yml: TIMEOUT_FACTOR: 1.
- Convert all 34 literal 'timeout:' task keywords / module args across 25
  task+handler files to "{{ lookup('timeout', <base>) | int }}".
- tests/lint/repository/yaml/test_timeout_uses_lookup.py enforces the
  lookup in every task/handler timeout (opt-out: # nocheck: timeout-lookup).
- swarm.yml pre-deploy build handler: add the missing timeout (same
  network-build hang vector as compose.yml's Build compose).
- Drop 13 pre-existing narration comments in the touched files (comment
  lint enforces whole-file on any edited file).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-03 13:57:30 +02:00

69 lines
2.5 KiB
Python

"""Scale a base timeout (seconds) by the global ``TIMEOUT_FACTOR``.
Replaces a bare ``timeout: <seconds>`` task keyword or module argument so every
wait in the codebase scales from one knob: set ``TIMEOUT_FACTOR: 2`` in
group_vars on a slow uplink and every timeout doubles.
timeout: "{{ lookup('timeout', 3600) | int }}"
timeout: "{{ lookup('timeout', some_base_var) | int }}"
``TIMEOUT_FACTOR`` is read from the play variables and defaults to 1 (no
scaling) when unset, so the lookup is a value-identical drop-in for the
literal. An explicit ``factor=`` keyword overrides the global for one call.
"""
from __future__ import annotations
from typing import Any
from ansible.errors import AnsibleError
from ansible.plugins.lookup import LookupBase
DEFAULT_TIMEOUT_FACTOR = 1
def scaled_timeout(base: Any, factor: Any = DEFAULT_TIMEOUT_FACTOR) -> int:
"""Return ``round(base * factor)`` as an int.
Args:
base: Base timeout in seconds (number or numeric string, >= 0).
factor: Multiplier (number or numeric string, >= 0); TIMEOUT_FACTOR.
Raises:
AnsibleError: base or factor is non-numeric or negative.
"""
try:
base_val = float(base)
except (TypeError, ValueError) as exc:
raise AnsibleError(
f"lookup('timeout', <base>): base {base!r} is not a number"
) from exc
try:
factor_val = float(factor)
except (TypeError, ValueError) as exc:
raise AnsibleError(
f"lookup('timeout'): TIMEOUT_FACTOR {factor!r} is not a number"
) from exc
if base_val < 0:
raise AnsibleError(f"lookup('timeout', <base>): base {base!r} must be >= 0")
if factor_val < 0:
raise AnsibleError(f"lookup('timeout'): TIMEOUT_FACTOR {factor!r} must be >= 0")
return round(base_val * factor_val)
class LookupModule(LookupBase):
"""lookup('timeout', base_seconds[, factor=TIMEOUT_FACTOR])"""
def run(self, terms, variables: dict[str, Any] | None = None, **kwargs):
terms = terms or []
if len(terms) != 1:
raise AnsibleError(
"lookup('timeout', base_seconds[, factor=]) expects exactly one "
"positional term (the base timeout in seconds)."
)
variables = variables or getattr(self._templar, "available_variables", {}) or {}
factor = kwargs.get("factor")
if factor is None:
factor = variables.get("TIMEOUT_FACTOR", DEFAULT_TIMEOUT_FACTOR)
return [scaled_timeout(terms[0], factor)]