Add automatic ascender_stats_* job artifacts (changed/failed flags and host lists)

awx/main/conf.py

@@ -340,6 +340,39 @@
     category_slug='jobs',
 )
 
+register(
+    'ASCENDER_AUTO_STATS_ENABLED',
+    field_class=fields.BooleanField,
+    default=True,
+    label=_('Enable Automatic Job Stats Artifacts'),
+    help_text=_(
+        'Automatically add ascender_stats_* keys (changed/failed flags and host name lists '
+        'derived from the playbook stats) to the artifacts of every finished playbook job '
+        '(jobs launched from job templates; project updates, inventory syncs and ad hoc '
+        'commands are not affected), so they can be used by downstream workflow nodes and '
+        'conditional connectors without requiring set_stats in the playbook. Can be '
+        'overridden per job or workflow with an ASCENDER_AUTO_STATS_ENABLED extra variable.'
+    ),
+    category=_('Jobs'),
+    category_slug='jobs',
+)
+
+register(
+    'ASCENDER_AUTO_STATS_MAX_HOSTS',
+    field_class=fields.IntegerField,
+    default=100,
+    min_value=0,
+    label=_('Maximum Hosts in Automatic Job Stats Artifacts'),
+    help_text=_(
+        'When a play involves more hosts than this, the ascender_stats_* host name lists are '
+        'omitted from the automatic job stats artifacts (the boolean flags are always kept) '
+        'and ascender_stats_hosts_truncated is set instead. Can be overridden per job or workflow '
+        'with an ASCENDER_AUTO_STATS_MAX_HOSTS extra variable.'
+    ),
+    category=_('Jobs'),
+    category_slug='jobs',
+)
+
 register(
     'AWX_COLLECTIONS_ENABLED',
     field_class=fields.BooleanField,

awx/main/models/workflow.py

@@ -142,6 +142,46 @@ def evaluate(self, artifacts):
         return evaluate_artifact_condition(artifacts, self.artifact_key, self.operator or 'eq', self.expected_value)
 
 
+# Automatic playbook stats artifacts, produced for every finished job
+# (see awx.main.tasks.callback.build_stats_artifacts). Since every job emits
+# the same key names, aggregating artifacts of several sibling jobs (slices of
+# a sliced job template, nodes of a nested workflow, converging parents) with
+# plain dict.update() would keep only the last job's values, hiding e.g. a
+# failed host reported by an earlier sibling. These keys are merged instead.
+STATS_BOOLEAN_ARTIFACTS = ('ascender_stats_changed', 'ascender_stats_failed', 'ascender_stats_hosts_truncated')
+STATS_HOST_LIST_ARTIFACTS = (
+    ('ascender_stats_changed_hosts', 'ascender_stats_non_changed_hosts'),
+    ('ascender_stats_failed_hosts', 'ascender_stats_non_failed_hosts'),
+)
+
+
+def merge_stats_artifacts(dest, src):
+    """Update dest with src like dict.update(), except the automatic
+    ascender_stats_* keys present on both sides are combined: booleans are OR-ed
+    and the host name lists are merged, so a host reported as changed or
+    failed by any of the aggregated jobs keeps that state. When any side was
+    truncated the merged host lists are dropped entirely, mirroring how
+    build_stats_artifacts never emits partial lists.
+    """
+    combined = {}
+    for key in STATS_BOOLEAN_ARTIFACTS:
+        if key in dest and key in src:
+            combined[key] = bool(dest[key]) or bool(src[key])
+    for positive_key, negative_key in STATS_HOST_LIST_ARTIFACTS:
+        if positive_key in dest and positive_key in src:
+            matched = set(dest.get(positive_key) or []) | set(src.get(positive_key) or [])
+            seen = matched | set(dest.get(negative_key) or []) | set(src.get(negative_key) or [])
+            combined[positive_key] = sorted(matched)
+            combined[negative_key] = sorted(seen - matched)
+    dest.update(src)
+    dest.update(combined)
+    if dest.get('ascender_stats_hosts_truncated'):
+        for positive_key, negative_key in STATS_HOST_LIST_ARTIFACTS:
+            dest.pop(positive_key, None)
+            dest.pop(negative_key, None)
+    return dest
+
+
 class WorkflowNodeBase(CreatedModifiedModel, LaunchTimeConfig):
     class Meta:
         abstract = True
@@ -479,9 +519,13 @@ def get_job_kwargs(self):
         is_root_node = True
         for parent_node in self.get_parent_nodes():
             is_root_node = False
-            aa_dict.update(parent_node.ancestor_artifacts)
+            # within one parent, its own job output supersedes what it inherited;
+            # across parents the stats keys are merged so one parent cannot mask
+            # the changed/failed hosts reported by another
+            parent_artifacts = dict(parent_node.ancestor_artifacts)
             if parent_node.job:
-                aa_dict.update(parent_node.job.get_effective_artifacts(parents_set=set([self.workflow_job_id])))
+                parent_artifacts.update(parent_node.job.get_effective_artifacts(parents_set=set([self.workflow_job_id])))
+            merge_stats_artifacts(aa_dict, parent_artifacts)
         if aa_dict and not is_root_node:
             self.ancestor_artifacts = aa_dict
             self.save(update_fields=['ancestor_artifacts'])
@@ -970,7 +1014,7 @@ def get_effective_artifacts(self, **kwargs):
         for job in job_queryset:
             if job.id in parents_set:
                 continue
-            artifacts.update(job.get_effective_artifacts(parents_set=new_parents_set))
+            merge_stats_artifacts(artifacts, job.get_effective_artifacts(parents_set=new_parents_set))
         return artifacts
 
     def prompts_dict(self, *args, **kwargs):

awx/main/tasks/callback.py

@@ -18,6 +18,62 @@
 logger = logging.getLogger('awx.main.tasks.callback')
 
 
+def _setting_as_bool(value, default):
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, int) and value in (0, 1):
+        return bool(value)
+    if isinstance(value, str):
+        normalized = value.strip().lower()
+        if normalized in ('true', 'yes', 'on', '1'):
+            return True
+        if normalized in ('false', 'no', 'off', '0'):
+            return False
+    return default
+
+
+def _setting_as_int(value, default):
+    if isinstance(value, bool):
+        return default
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+def build_stats_artifacts(stats_event_data, max_hosts):
+    """Build the ascender_stats_* artifact entries from a playbook_on_stats event.
+
+    The host name lists are omitted (and ascender_stats_hosts_truncated set) when the
+    play involved more than max_hosts hosts, so artifacts stay reasonably small
+    as they propagate to descendant workflow nodes.
+    """
+    counts = {}
+    for stat in ('changed', 'failures', 'dark', 'ok', 'processed', 'skipped', 'ignored', 'rescued'):
+        value = stats_event_data.get(stat)
+        counts[stat] = value if isinstance(value, dict) else {}
+    all_hosts = set()
+    for host_counts in counts.values():
+        all_hosts.update(host_counts.keys())
+    changed_hosts = {host for host in all_hosts if counts['changed'].get(host)}
+    failed_hosts = {host for host in all_hosts if counts['failures'].get(host) or counts['dark'].get(host)}
+    stats_artifacts = {
+        'ascender_stats_changed': bool(changed_hosts),
+        'ascender_stats_failed': bool(failed_hosts),
+        'ascender_stats_hosts_truncated': len(all_hosts) > max_hosts,
+    }
+    if not stats_artifacts['ascender_stats_hosts_truncated']:
+        stats_artifacts.update(
+            {
+                'ascender_stats_changed_hosts': sorted(changed_hosts),
+                'ascender_stats_non_changed_hosts': sorted(all_hosts - changed_hosts),
+                'ascender_stats_failed_hosts': sorted(failed_hosts),
+                'ascender_stats_non_failed_hosts': sorted(all_hosts - failed_hosts),
+            }
+        )
+    return stats_artifacts
+
+
 class RunnerCallback:
     def __init__(self, model=None):
         self.parent_workflow_job_id = None
@@ -170,11 +226,37 @@ def event_handler(self, event_data):
         '''
         Handle artifacts
         '''
-        if event_data.get('event_data', {}).get('artifact_data', {}):
-            self.delay_update(artifacts=event_data['event_data']['artifact_data'])
+        artifact_data = event_data.get('event_data', {}).get('artifact_data', {})
+        if event_data.get('event', '') == 'playbook_on_stats' and self.event_data_key == 'job_id':
+            stats_artifacts = self.get_stats_artifacts(event_data.get('event_data', {}))
+            if stats_artifacts:
+                # set_stats data provided by the playbook wins over the automatic keys
+                stats_artifacts.update(artifact_data)
+                artifact_data = stats_artifacts
+        if artifact_data:
+            self.delay_update(artifacts=artifact_data)
 
         return False
 
+    def get_stats_artifacts(self, stats_event_data):
+        """Return the automatic ascender_stats_* artifacts for the final stats event,
+        or an empty dict when the feature is disabled. This only runs once per
+        job, so reading the job extra_vars here does not affect the hot path
+        warned about in event_handler."""
+        try:
+            job_extra_vars = self.instance.extra_vars_dict
+        except Exception:
+            logger.exception(f'Failed to parse extra_vars of {self.instance.log_format}, using defaults for automatic stats artifacts')
+            job_extra_vars = {}
+        if not _setting_as_bool(job_extra_vars.get('ASCENDER_AUTO_STATS_ENABLED'), settings.ASCENDER_AUTO_STATS_ENABLED):
+            return {}
+        max_hosts = _setting_as_int(job_extra_vars.get('ASCENDER_AUTO_STATS_MAX_HOSTS'), settings.ASCENDER_AUTO_STATS_MAX_HOSTS)
+        if max_hosts < 0:
+            # the setting itself is validated with min_value=0, but the extra_vars
+            # override is not; a negative value would flag every play as truncated
+            max_hosts = settings.ASCENDER_AUTO_STATS_MAX_HOSTS
+        return build_stats_artifacts(stats_event_data, max_hosts)
+
     def finished_callback(self, runner_obj):
         """
         Ansible runner callback triggered on finished run

awx/main/tests/unit/models/test_workflow_unit.py

@@ -2,7 +2,7 @@
 
 from awx.main.models.jobs import JobTemplate
 from awx.main.models import Inventory, CredentialType, Credential, Project
-from awx.main.models.workflow import WorkflowJobTemplate, WorkflowJobTemplateNode, WorkflowJob, WorkflowJobNode
+from awx.main.models.workflow import WorkflowJobTemplate, WorkflowJobTemplateNode, WorkflowJob, WorkflowJobNode, merge_stats_artifacts
 from unittest import mock
 
 
@@ -245,3 +245,62 @@ def test_get_ask_mapping_integrity():
         'skip_tags',
         'extra_vars',
     ]
+
+
+class TestMergeStatsArtifacts:
+    """merge_stats_artifacts combines the automatic ascender_stats_* keys across
+    sibling jobs (slices, nested workflow nodes, converging parents) instead of
+    letting the last finished job overwrite the others."""
+
+    def test_plain_keys_keep_update_semantics(self):
+        dest = {'custom': 1, 'other': 'a'}
+        merge_stats_artifacts(dest, {'custom': 2})
+        assert dest == {'custom': 2, 'other': 'a'}
+
+    def test_booleans_are_ored_across_jobs(self):
+        # slice 1 failed a host, slice 2 was clean and finished last
+        dest = {'ascender_stats_changed': True, 'ascender_stats_failed': True, 'ascender_stats_hosts_truncated': False}
+        merge_stats_artifacts(dest, {'ascender_stats_changed': False, 'ascender_stats_failed': False, 'ascender_stats_hosts_truncated': False})
+        assert dest['ascender_stats_changed'] is True
+        assert dest['ascender_stats_failed'] is True
+        assert dest['ascender_stats_hosts_truncated'] is False
+
+    def test_host_lists_are_unioned(self):
+        dest = {
+            'ascender_stats_changed_hosts': ['h1'],
+            'ascender_stats_non_changed_hosts': ['h2'],
+            'ascender_stats_failed_hosts': [],
+            'ascender_stats_non_failed_hosts': ['h1', 'h2'],
+        }
+        src = {
+            'ascender_stats_changed_hosts': ['h2', 'h3'],
+            'ascender_stats_non_changed_hosts': ['h1'],
+            'ascender_stats_failed_hosts': ['h3'],
+            'ascender_stats_non_failed_hosts': ['h1', 'h2'],
+        }
+        merge_stats_artifacts(dest, src)
+        # a host that changed (or failed) in any job stays in the positive list
+        assert dest['ascender_stats_changed_hosts'] == ['h1', 'h2', 'h3']
+        assert dest['ascender_stats_non_changed_hosts'] == []
+        assert dest['ascender_stats_failed_hosts'] == ['h3']
+        assert dest['ascender_stats_non_failed_hosts'] == ['h1', 'h2']
+
+    def test_truncation_drops_lists(self):
+        dest = {
+            'ascender_stats_hosts_truncated': False,
+            'ascender_stats_changed_hosts': ['h1'],
+            'ascender_stats_non_changed_hosts': [],
+        }
+        merge_stats_artifacts(dest, {'ascender_stats_hosts_truncated': True})
+        assert dest['ascender_stats_hosts_truncated'] is True
+        assert 'ascender_stats_changed_hosts' not in dest
+        assert 'ascender_stats_non_changed_hosts' not in dest
+
+    def test_one_sided_keys_are_preserved(self):
+        # a sibling without stats (feature disabled, non-playbook job) must not
+        # erase what another sibling reported
+        dest = {'ascender_stats_failed': True, 'ascender_stats_failed_hosts': ['h1'], 'ascender_stats_non_failed_hosts': []}
+        merge_stats_artifacts(dest, {'some_set_stats_key': 'x'})
+        assert dest['ascender_stats_failed'] is True
+        assert dest['ascender_stats_failed_hosts'] == ['h1']
+        assert dest['some_set_stats_key'] == 'x'