Read changed hosts from the play recap, not the last task

changelog.d/76-changed-from-play-stats.md

@@ -0,0 +1,10 @@
+### Fixed
+
+- A decision records a host `changed` when any task of the play changed it,
+  not only the last one. The check preview, the execution and the verify run
+  read each host's counts from the play recap through a callback the SDK
+  ships (`cloudfall_host_stats`), in place of Ansible's `tree` callback,
+  which keeps only the last task's result: a run whose recap said
+  `changed=2` was recorded `changed: []`, and a verify run that changed a
+  host was recorded `verified`. The callback is added ahead of the callback
+  directories the team's `ansible.cfg` names, not in place of them (#76)

sdk/src/cloudfall/ansible_api.py

@@ -10,6 +10,9 @@
 function falls back to the ``ansible-inventory`` command and says so in
 ``AnsibleInventory.in_process``, so a broken internal API degrades to a
 slower read rather than to no read at all.
+
+The callback plugin in ``cloudfall.callback_plugins`` imports ansible-core
+too, but Ansible loads it into ``ansible-playbook``; the SDK never does.
 """
 
 from __future__ import annotations
@@ -208,6 +211,34 @@ def inventory_from_config(directory: Path) -> InventorySource | None:
     return InventorySource(candidate)
 
 
+def callback_plugin_path(configuration: Path | None) -> tuple[Path, ...]:
+    """Return the callback plugin directories a playbook run would search.
+
+    ``ANSIBLE_CALLBACK_PLUGINS`` replaces the configured path rather than
+    adding to it, so a caller that adds a directory of its own starts from
+    this: the operator's environment, else the team's ``ansible.cfg`` (its
+    relative entries resolved against the file), else Ansible's default.
+    ``None`` searches for a configuration the way ``ansible-playbook`` does.
+    """
+    from ansible.config.manager import (  # noqa: PLC0415 - lazy by design
+        ConfigManager,
+    )
+    from ansible.errors import AnsibleError  # noqa: PLC0415 - lazy by design
+
+    try:
+        manager = ConfigManager(
+            str(configuration.resolve()) if configuration is not None else None
+        )
+        home = manager.get_config_value("ANSIBLE_HOME")
+        paths = manager.get_config_value(
+            "DEFAULT_CALLBACK_PLUGIN_PATH", variables={"ANSIBLE_HOME": home}
+        )
+    except AnsibleError as error:
+        message = f"the callback plugin path is not readable: {error}"
+        raise AnsibleReadError(_ERROR_CONFIG_UNREADABLE, message) from error
+    return tuple(Path(str(entry)) for entry in paths)
+
+
 def read_inventory(
     source: InventorySource, *, limit: str | None = None, in_process: bool = True
 ) -> AnsibleInventory:

sdk/src/cloudfall/callback_plugins/__init__.py

@@ -0,0 +1,6 @@
+"""Ansible callback plugins the SDK enables on the playbooks it runs.
+
+Ansible, not the SDK, imports what is in this directory: ``run_command``
+points ``ANSIBLE_CALLBACK_PLUGINS`` at it. Ansible skips files whose name
+starts with an underscore, so this one is never loaded as a plugin.
+"""

sdk/src/cloudfall/callback_plugins/cloudfall_host_stats.py

@@ -0,0 +1,65 @@
+"""Each host's play recap, one JSON file per host, for the decision record.
+
+Whether a run changed a host is a fact of the whole play, not of its last
+task, so it is read from the play stats the recap prints rather than from
+per-task results. Ansible loads this file itself, into ``ansible-playbook``;
+it imports nothing from the SDK, so it works whichever interpreter runs it.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+from ansible.plugins.callback import CallbackBase
+
+if TYPE_CHECKING:
+    from ansible.executor.stats import AggregateStats
+
+DOCUMENTATION = """
+name: cloudfall_host_stats
+type: aggregate
+short_description: Write each host's play recap counts as JSON
+description:
+  - When the playbook ends, writes one JSON file per host, named after the
+    host, holding its ok, changed, failed and unreachable counts.
+requirements:
+  - enabled in callbacks_enabled
+options:
+  directory:
+    description: The directory the per-host files are written to.
+    required: true
+    type: path
+    env:
+      - name: CLOUDFALL_HOST_STATS_DIR
+"""
+
+# What each written key is called in the stats' own summary.
+_COUNTS = {
+    "ok": "ok",
+    "changed": "changed",
+    "failed": "failures",
+    "unreachable": "unreachable",
+}
+
+
+# ansible-core ships no type information, so its base class is Any to mypy.
+class CallbackModule(CallbackBase):  # type: ignore[misc]
+    """Write the recap counts of every host the play touched."""
+
+    CALLBACK_VERSION = 2.0
+    CALLBACK_TYPE = "aggregate"
+    CALLBACK_NAME = "cloudfall_host_stats"
+    CALLBACK_NEEDS_ENABLED = True
+
+    def v2_playbook_on_stats(self, stats: AggregateStats) -> None:
+        """Write one file per host once the play's stats are final."""
+        directory = Path(str(self.get_option("directory")))
+        directory.mkdir(parents=True, exist_ok=True)
+        for host in sorted(stats.processed):
+            summary = stats.summarize(host)
+            counts = {key: int(summary[name]) for key, name in _COUNTS.items()}
+            (directory / host).write_text(
+                json.dumps(counts, sort_keys=True), encoding="utf-8"
+            )

sdk/src/cloudfall/decision.py

@@ -29,6 +29,7 @@
 
 from jsonschema.exceptions import ValidationError
 
+from cloudfall.ansible_api import callback_plugin_path
 from cloudfall.catalog import RiskLevel, TargetScope
 from cloudfall.domain import ResourceId
 
@@ -54,11 +55,19 @@
 matters.
 """
 
-_TREE_DIRECTORY = "tree"
+_HOST_STATS_DIRECTORY = "host-stats"
+HOST_STATS_CALLBACK = "cloudfall_host_stats"
+"""The callback that leaves each host's recap counts; see cloudfall.callback_plugins."""
+
+HOST_STATS_VARIABLE = "CLOUDFALL_HOST_STATS_DIR"
+"""Where that callback writes one JSON file per host."""
+
+_CALLBACK_PLUGIN_DIRECTORY = Path(__file__).parent / "callback_plugins"
 _DIFF_SUFFIX = ".diff"
 _ANSIBLE_CONFIG_FILE = "ansible.cfg"
 CHECK_TIMEOUT_SECONDS = 1800
 _CHANGED_KEY = "changed"
+_ERROR_HOST_STATS_INVALID = "decision_host_stats_invalid"
 
 _ERROR_ANSIBLE_MISSING = "decision_ansible_missing"
 _ERROR_TARGET_REQUIRED = "decision_target_required"
@@ -460,12 +469,12 @@ def propose(
     moment = (now or _utc_now)()
     decision_id = _decision_identifier(operation.operation_id, moment)
     diff_path = store.directory / f"{decision_id.value}{_DIFF_SUFFIX}"
-    tree = store.directory / _TREE_DIRECTORY / decision_id.value
+    host_stats = store.directory / _HOST_STATS_DIRECTORY / decision_id.value
     argv, environment = run_command(
         PlaybookInvocation(
             playbook=operation.playbook,
             repository=request.repository,
-            tree=tree,
+            host_stats=host_stats,
             pattern=pattern,
             inputs=declared,
         ),
@@ -489,7 +498,7 @@ def propose(
         requirement=requirement,
         reason=reason,
         basis=read_basis(request.observations, request.repository),
-        check=_preview(exit_code, tree, diff_path, request.repository),
+        check=_preview(exit_code, host_stats, diff_path, request.repository),
         status=DecisionStatus.RAN if ran else DecisionStatus.PROPOSED,
         verdict=_ran_verdict(exit_code, request.replay) if ran else None,
         ran_exit_code=exit_code if ran else None,
@@ -533,7 +542,9 @@ class PlaybookInvocation:
 
     playbook: Path
     repository: Path
-    tree: Path
+    host_stats: Path
+    """Where the run leaves each host's recap counts, one JSON file per host."""
+
     pattern: str | None = None
     inputs: Mapping[str, object] = field(default_factory=dict)
 
@@ -711,15 +722,17 @@ def _run_stage(
     invocation = PlaybookInvocation(
         playbook=playbook,
         repository=stage.request.repository,
-        tree=stage.store.directory / _TREE_DIRECTORY / f"{identifier}-{name}",
+        host_stats=(
+            stage.store.directory / _HOST_STATS_DIRECTORY / f"{identifier}-{name}"
+        ),
         pattern=decision.targets.pattern,
         inputs=decision.inputs,
     )
     argv, environment = run_command(invocation)
     log_path.parent.mkdir(parents=True, exist_ok=True)
     exit_code = stage.runner(argv, environment, log_path)
     preview = _preview(
-        exit_code, invocation.tree, log_path, stage.request.repository
+        exit_code, invocation.host_stats, log_path, stage.request.repository
     )
     return RunRecord(
         ran_at=_timestamp(moment),
@@ -737,7 +750,9 @@ def run_command(
 
     The team's own configuration decides how Ansible connects and where
     their roles live, because this is their playbook: nothing about the
-    engine belongs in this run.
+    engine belongs in this run. The one thing added is the callback that
+    leaves each host's recap counts, put ahead of the callback directories
+    their configuration already names rather than in place of them.
     """
     binary = shutil.which("ansible-playbook")
     if binary is None:
@@ -751,13 +766,16 @@ def run_command(
             ("--extra-vars", json.dumps(dict(invocation.inputs), sort_keys=True))
         )
     argv.append(str(invocation.repository / invocation.playbook))
+    configuration = invocation.repository / _ANSIBLE_CONFIG_FILE
+    team = configuration if configuration.is_file() else None
+    plugins = (_CALLBACK_PLUGIN_DIRECTORY, *callback_plugin_path(team))
     environment = {
-        "ANSIBLE_CALLBACKS_ENABLED": _TREE_DIRECTORY,
-        "ANSIBLE_CALLBACK_TREE_DIR": str(invocation.tree.resolve()),
+        "ANSIBLE_CALLBACK_PLUGINS": os.pathsep.join(str(path) for path in plugins),
+        "ANSIBLE_CALLBACKS_ENABLED": HOST_STATS_CALLBACK,
+        HOST_STATS_VARIABLE: str(invocation.host_stats.resolve()),
     }
-    configuration = invocation.repository / _ANSIBLE_CONFIG_FILE
-    if configuration.is_file():
-        environment["ANSIBLE_CONFIG"] = str(configuration.resolve())
+    if team is not None:
+        environment["ANSIBLE_CONFIG"] = str(team.resolve())
     return tuple(argv), environment
 
 
@@ -801,23 +819,20 @@ def read_basis(observations: Path, repository: Path | None = None) -> Basis:
 
 
 def _preview(
-    exit_code: int, tree: Path, diff_path: Path, repository: Path | None = None
+    exit_code: int, host_stats: Path, diff_path: Path, repository: Path | None = None
 ) -> CheckPreview:
+    """Sort the hosts by what the whole play did to them, then cite the output.
+
+    A host is changed when any task changed it: the recap's count says so,
+    where the last task's result would only say what that one task did.
+    """
     changed: list[str] = []
     unchanged: list[str] = []
-    if tree.is_dir():
-        for path in sorted(tree.iterdir()):
+    if host_stats.is_dir():
+        for path in sorted(host_stats.iterdir()):
             if not path.is_file():
                 continue
-            document = cast(
-                "object", json.loads(path.read_text(encoding="utf-8"))
-            )
-            reported = (
-                bool(document.get(_CHANGED_KEY))
-                if isinstance(document, dict)
-                else False
-            )
-            (changed if reported else unchanged).append(path.name)
+            (changed if _changed_count(path) > 0 else unchanged).append(path.name)
     content = diff_path.read_bytes() if diff_path.is_file() else b""
     return CheckPreview(
         exit_code=exit_code,
@@ -835,6 +850,16 @@ def _preview(
     )
 
 
+def _changed_count(path: Path) -> int:
+    """Return how many tasks changed the host one recap file names."""
+    document = cast("object", json.loads(path.read_text(encoding="utf-8")))
+    count = document.get(_CHANGED_KEY) if isinstance(document, dict) else None
+    if not isinstance(count, int) or isinstance(count, bool):
+        message = f"{path} holds no {_CHANGED_KEY} count of a host's recap"
+        raise DecisionError(_ERROR_HOST_STATS_INVALID, message)
+    return count
+
+
 def _validated_targets(operation: Operation, targets: Targets) -> str | None:
     if targets.scope is not operation.targets:
         message = (

sdk/src/cloudfall/recording.py

@@ -18,6 +18,8 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, cast
 
+from cloudfall.decision import HOST_STATS_VARIABLE
+
 if TYPE_CHECKING:
     from collections.abc import Mapping, Sequence
 
@@ -29,8 +31,6 @@
 ERROR_RECORDING_EMPTY = "recording_empty"
 """The code of a recording that is not a directory or holds no runs."""
 
-_TREE_DIR_VARIABLE = "ANSIBLE_CALLBACK_TREE_DIR"
-
 
 class RecordingError(ValueError):
     """A recording that cannot be played back."""
@@ -118,7 +118,7 @@ def run(argv: Sequence[str], environment: Mapping[str, str], diff: Path) -> int:
                 )
                 return NOT_RECORDED_EXIT
             diff.write_text(recorded.output, encoding="utf-8")
-            _write_tree(environment, recorded)
+            _write_host_stats(environment, recorded)
             return recorded.exit_code
 
         return run
@@ -150,14 +150,20 @@ def _value(arguments: Sequence[str], flag: str) -> str | None:
     return arguments[arguments.index(flag) + 1]
 
 
-def _write_tree(environment: Mapping[str, str], recorded: RecordedRun) -> None:
-    """Leave the per-host results the tree callback would have written."""
-    tree = environment.get(_TREE_DIR_VARIABLE)
-    if tree is None:
+def _write_host_stats(environment: Mapping[str, str], recorded: RecordedRun) -> None:
+    """Leave the per-host recap counts the host stats callback would have written.
+
+    A recording keeps which hosts changed, not how many tasks did, so each
+    host is played back as one task: changed, or ok and unchanged.
+    """
+    named = environment.get(HOST_STATS_VARIABLE)
+    if named is None:
         return
-    directory = Path(tree)
+    directory = Path(named)
     directory.mkdir(parents=True, exist_ok=True)
-    for host in recorded.changed:
-        (directory / host).write_text('{"changed": true}', encoding="utf-8")
-    for host in recorded.unchanged:
-        (directory / host).write_text('{"changed": false}', encoding="utf-8")
+    for hosts, changed in ((recorded.changed, 1), (recorded.unchanged, 0)):
+        for host in hosts:
+            counts = {"ok": 1, "changed": changed, "failed": 0, "unreachable": 0}
+            (directory / host).write_text(
+                json.dumps(counts, sort_keys=True), encoding="utf-8"
+            )