chore(ci): extend argspec drift check with a bare is-defined scanner

CLAUDE.md

@@ -13,3 +13,26 @@ When fixing a bug:
    genuinely requires it (e.g. inter-node communication).
 4. The test should fail without the fix and pass with it. Confirm this mentally
    or by describing the failure mode before implementing.
+
+## Role default gating
+
+When a role default is commented out (`# foo:`) or has an empty value
+(`foo:`, `foo: ""`, `foo: []`, `foo: {}`), the downstream gate must be
+an explicit non-empty check, not a bare `is defined`:
+
+```yaml
+# Wrong — silently changes behavior when someone uncomments foo:
+when: foo is defined
+
+# Right — the empty sentinel behaves the same whether the default is
+# commented, missing, or explicitly ""
+when: foo | default('') | length > 0
+```
+
+The `elasticstack_cert_pass: ""` regression that took several rounds
+to diagnose (empty-string propagated into `elasticsearch-keystore add`
+stdin and `elasticsearch-certutil --pass`) is the archetypal case.
+
+`scripts/check_argspecs.py` scans role tasks for this pattern and
+fails CI when a bare `is defined` gate references a default var whose
+declared value is empty or null.

roles/beats/tasks/metricbeat.yml

@@ -23,8 +23,8 @@
   ansible.builtin.command: "/usr/bin/metricbeat modules enable {{ item }}"
   args:
     creates: "/etc/metricbeat/modules.d/{{ item }}.yml"
-  loop: "{{ beats_metricbeat_modules }}"
-  when: beats_metricbeat_modules is defined
+  loop: "{{ beats_metricbeat_modules | default([], true) }}"
+  when: beats_metricbeat_modules | default([], true) | length > 0
 
 - name: metricbeat | Enable Ingest Pipelines
   ansible.builtin.shell: >
@@ -37,7 +37,7 @@
   notify:
     - Restart Metricbeat
   when:
-    - beats_metricbeat_modules is defined
+    - beats_metricbeat_modules | default([], true) | length > 0
     - beats_metricbeat_output == "elasticsearch"
 
 - name: metricbeat | Start Metricbeat

roles/elasticsearch/tasks/main.yml

@@ -395,7 +395,6 @@
     _elasticsearch_extra_config_conflicts: >-
       {{ elasticsearch_extra_config.keys() | list | intersect(_elasticsearch_managed_keys) }}
   when:
-    - elasticsearch_extra_config is defined
     - _elasticsearch_extra_config_conflicts | length > 0
 
 - name: Configure Elasticsearch

scripts/check_argspecs.py

@@ -43,6 +43,37 @@ def defaults_vars(path):
     return out
 
 
+def defaults_empty_vars(path):
+    """Names whose value in defaults/main.yml is empty or null.
+
+    An entry that looks like `foo:` (nothing after the colon) or
+    `foo: ""` / `foo: []` / `foo: {}` / `foo: null` counts as empty.
+    Those are the ones where `is defined` in downstream gates silently
+    changes meaning the moment somebody supplies an empty string —
+    which is exactly the elasticstack_cert_pass regression class.
+    """
+    out = set()
+    empty_val = re.compile(
+        r"""^([a-z][\w]*)\s*:\s*
+            (?:                       # then either:
+                (?:\#.*)?             # trailing comment only
+                |""                   # empty string
+                |''
+                |\[\s*\]              # empty list
+                |\{\s*\}              # empty dict
+                |[Nn]ull
+                |~
+            )?\s*(?:\#.*)?$""",
+        re.VERBOSE,
+    )
+    with open(path) as f:
+        for line in f:
+            m = empty_val.match(line)
+            if m:
+                out.add(m.group(1))
+    return out
+
+
 def argspec_options(path):
     """Option names declared under the role's `main` entry point.
 
@@ -58,10 +89,107 @@ def argspec_options(path):
     return set((main.get("options") or {}).keys())
 
 
+IS_DEFINED_RE = re.compile(r"\b([a-z][\w]*)\s+is\s+defined\b")
+
+
+def _paired_guard_for(var, expr):
+    """Return True iff `expr` also contains a same-var guard for `var`
+    that establishes non-emptiness or register-shape existence.
+
+    Guards recognised (all with `var` as the target, no other names):
+
+      - `var | ... | length ...`  (any filter chain that ends in `length`)
+      - `var.stdout` / `var.content`  (register reference)
+      - `var | ... | bool`  (truthy check — `""` is falsey)
+
+    The check is targeted at `var` specifically. An unrelated
+    `other.stdout` or `other | length` in the same expression does NOT
+    excuse a bare `var is defined`.
+    """
+    var_re = re.escape(var)
+    # `var` followed by any pipe-chain that ultimately reaches `length`
+    # or `bool` — matches `var | length`, `var | string | length > 0`,
+    # `var | default('') | length > 0`, `var | bool`, etc. Intermediate
+    # filters can be `name(args)` or bare names.
+    filter_chain = r"(?:\s*\|\s*\w+(?:\([^)]*\))?)*"
+    patterns = (
+        rf"\b{var_re}{filter_chain}\s*\|\s*length\b",
+        rf"\b{var_re}{filter_chain}\s*\|\s*bool\b",
+        rf"\b{var_re}\.(?:stdout|content)\b",
+    )
+    return any(re.search(p, expr) for p in patterns)
+
+
+def _flatten_when(when):
+    """Ansible's `when` is either a string, a list of strings, or a
+    boolean. Yield each condition-string; ignore booleans."""
+    if isinstance(when, str):
+        yield when
+    elif isinstance(when, list):
+        for item in when:
+            if isinstance(item, str):
+                yield item
+    # bool / None / dict → nothing to check
+
+
+def _walk_tasks(items, path, empties_for_role, hits):
+    """Recurse through a task list, inspecting each task's `when:`
+    (and any nested block/rescue/always children)."""
+    if not isinstance(items, list):
+        return
+    for item in items:
+        if not isinstance(item, dict):
+            continue
+
+        name = str(item.get("name", ""))
+        for cond in _flatten_when(item.get("when")):
+            for m in IS_DEFINED_RE.finditer(cond):
+                var = m.group(1)
+                if var not in empties_for_role:
+                    continue
+                if _paired_guard_for(var, cond):
+                    continue
+                hits.append((path, name, var, cond.strip()))
+
+        for sub_key in ("block", "rescue", "always"):
+            if sub_key in item:
+                _walk_tasks(item[sub_key], path, empties_for_role, hits)
+
+
+def scan_is_defined_gates(repo, empty_default_vars_by_role):
+    """Flag bare `X is defined` gates in role tasks where X is a
+    same-role default var whose declared value is empty or null.
+
+    Task YAML is parsed and only `when:` expressions are inspected so
+    comments and task names never trigger a false positive. The empty
+    defaults set is scoped per role so an empty `foo` in role A does
+    not falsely flag `foo is defined` in role B (where `foo` may hold
+    a real default). A `X is defined` is exempted only when the same
+    condition also contains a same-var non-empty guard
+    (`X | length …`, `X | bool`, `X.stdout`, `X.content`, or
+    `X | default(...) | length …`).
+    """
+    hits = []
+    for role_dir in sorted((repo / "roles").iterdir()):
+        if not role_dir.is_dir():
+            continue
+        empties = empty_default_vars_by_role.get(role_dir.name, set())
+        if not empties:
+            continue
+        for task_file in sorted((role_dir / "tasks").rglob("*.yml")):
+            try:
+                loaded = yaml.safe_load(task_file.read_text()) or []
+            except (OSError, yaml.YAMLError):
+                continue
+            _walk_tasks(loaded, task_file.relative_to(repo), empties, hits)
+    return hits
+
+
 def main():
     repo = Path(__file__).resolve().parent.parent
     roles = sorted(p for p in (repo / "roles").iterdir() if p.is_dir())
     exit_code = 0
+    empty_default_vars_by_role = {}
 
     for role in roles:
         defaults = role / "defaults" / "main.yml"
@@ -77,6 +205,7 @@ def main():
             continue
 
         vars_ = defaults_vars(defaults)
+        empty_default_vars_by_role[role.name] = defaults_empty_vars(defaults)
         opts = argspec_options(specs)
         missing_in_spec = vars_ - opts
         extra_in_spec = opts - vars_
@@ -91,9 +220,24 @@ def main():
         else:
             print(f"[{role.name}] ok ({len(vars_)} vars)")
 
+    hits = scan_is_defined_gates(repo, empty_default_vars_by_role)
+    if hits:
+        exit_code = 1
+        print("\nBare `X is defined` gates on role default vars:", file=sys.stderr)
+        print(
+            "These silently change behavior when the default is uncommented "
+            "or set to an empty value (see the elasticstack_cert_pass regression "
+            "for the pattern this catches). Rewrite the gate to check for "
+            "non-empty explicitly, e.g. `X | default('') | length > 0`.",
+            file=sys.stderr,
+        )
+        for path, task_name, var, cond in hits:
+            label = task_name or "(unnamed task)"
+            print(f"  {path} — {label}: `{cond}` (var: {var})", file=sys.stderr)
+
     if exit_code:
         print(
-            "\nFix by editing meta/argument_specs.yml or regenerating with "
+            "\nFix drift by editing meta/argument_specs.yml or regenerating with "
             "`scripts/gen_argspecs.py <role_path>`.",
             file=sys.stderr,
         )