From eb7090dd0f93d48d984c835639834979b5b17028 Mon Sep 17 00:00:00 2001 From: Stevenson Michel Date: Tue, 25 Aug 2026 14:55:43 -0400 Subject: [PATCH] [Devel] Fix Indirect Node Audit Null Top Level Name (#16610) * Added null guard and restructured transactions * Flag names that are None and add logging warnings for all modules * Add test for job-deleted-during-transaction race condition Covers the Job.DoesNotExist handler inside the transaction.atomic() block to satisfy SonarQube's 90% new code coverage gate. Co-Authored-By: Claude Opus 4.6 (1M context) --------- Co-authored-by: Claude Opus 4.6 (1M context) --- awx/main/tasks/host_indirect.py | 44 ++++++------ .../functional/tasks/test_host_indirect.py | 70 +++++++++++++++++++ 2 files changed, 92 insertions(+), 22 deletions(-) diff --git a/awx/main/tasks/host_indirect.py b/awx/main/tasks/host_indirect.py index e271021962..1d7f8919d6 100644 --- a/awx/main/tasks/host_indirect.py +++ b/awx/main/tasks/host_indirect.py @@ -44,6 +44,7 @@ def build_indirect_host_data(job: Job, job_event_queries: dict[str, dict[str, st compiled_jq_expressions = {} # Cache for compiled jq expressions facts_missing_logged = False unhashable_facts_logged = False + name_missing_logged = set() job_event_queries_fqcn = {} for query_k, query_v in job_event_queries.items(): @@ -103,6 +104,11 @@ def build_indirect_host_data(job: Job, job_event_queries: dict[str, dict[str, st # Obtain the record based on the hashable canonical_facts now determined facts = data.get('facts') name = data.get('name') + if name is None: + if resolved_action not in name_missing_logged: + logger.warning(f'jq output missing name for module {resolved_action} on event {event.id} using jq:{jq_str_for_event}') + name_missing_logged.add(resolved_action) + continue if hashable_facts in results: audit_record = results[hashable_facts] @@ -147,7 +153,6 @@ def save_indirect_host_entries_of_job(job: Job) -> None: job_event_queries = fetch_job_event_query(job) records = build_indirect_host_data(job, job_event_queries) IndirectManagedNodeAudit.objects.bulk_create(records) - job.event_queries_processed = True def cleanup_old_indirect_host_entries() -> None: @@ -175,28 +180,23 @@ def save_indirect_host_entries(job_id: int, wait_for_events: bool = True) -> Non return job.log_lifecycle(f'finished processing {current_events} events, running save_indirect_host_entries') - with transaction.atomic(): - """ - Pre-emptively set the job marker to 'events processed'. This prevents other instances from running the - same task. - """ - try: - job = Job.objects.select_for_update().get(id=job_id) - except job.DoesNotExist: - logger.debug(f'Job {job_id} seems to be deleted, bailing from save_indirect_host_entries') - return - - if job.event_queries_processed is True: - # this can mean one of two things: - # 1. another instance has already processed the events of this job - # 2. the artifacts_handler has not yet been called for this job - return - - job.event_queries_processed = True - job.save(update_fields=['event_queries_processed']) - try: - save_indirect_host_entries_of_job(job) + with transaction.atomic(): + try: + job = Job.objects.select_for_update().get(id=job_id) + except Job.DoesNotExist: + logger.debug(f'Job {job_id} seems to be deleted, bailing from save_indirect_host_entries') + return + + if job.event_queries_processed is True: + # this can mean one of two things: + # 1. another instance has already processed the events of this job + # 2. the artifacts_handler has not yet been called for this job + return + + save_indirect_host_entries_of_job(job) + job.event_queries_processed = True + job.save(update_fields=['event_queries_processed']) except Exception: logger.exception(f'Error processing indirect host data for job_id={job_id}') diff --git a/awx/main/tests/functional/tasks/test_host_indirect.py b/awx/main/tests/functional/tasks/test_host_indirect.py index cfa98d2391..798d6cbe4f 100644 --- a/awx/main/tests/functional/tasks/test_host_indirect.py +++ b/awx/main/tests/functional/tasks/test_host_indirect.py @@ -4,8 +4,11 @@ from unittest import mock import pytest +from django.db import transaction from django.utils.timezone import now, timedelta +from awx.main.models import Job + from awx.main.tasks.host_indirect import ( build_indirect_host_data, fetch_job_event_query, @@ -364,3 +367,70 @@ def test_cleanup_old_audit_records(old_audit_record, new_audit_record): cleanup_and_save_indirect_host_entries_fallback() count_after_cleanup = IndirectManagedNodeAudit.objects.count() assert count_after_cleanup == 1 + + +@mock.patch('awx.main.tasks.host_indirect.logger.warning') +@pytest.mark.django_db +def test_null_name_event_skipped_with_warning(mock_logger_warning, bare_job, event_query): + """Event with name=None should be skipped and a warning logged once.""" + bare_job.job_events.create( + event_data={ + 'resolved_action': 'demo.query.example', + 'res': {'direct_host_name': 'foo_host', 'name': None}, + } + ) + query = Query('demo.query.example', TEST_JQ) + records = build_indirect_host_data(bare_job, query) + assert records == [] + mock_logger_warning.assert_called_once() + assert 'missing name' in mock_logger_warning.call_args[0][0] + + +@pytest.mark.django_db +def test_mixed_valid_and_null_name_events(bare_job, event_query): + """Job with one valid and one null-name event: valid record persists, null-name is skipped.""" + bare_job.job_events.create( + event_data={ + 'resolved_action': 'demo.query.example', + 'res': {'direct_host_name': 'foo_host', 'name': 'vm-foo'}, + } + ) + bare_job.job_events.create( + event_data={ + 'resolved_action': 'demo.query.example', + 'res': {'direct_host_name': 'bar_host', 'name': None}, + } + ) + save_indirect_host_entries(bare_job.id, wait_for_events=False) + bare_job.refresh_from_db() + + assert bare_job.event_queries_processed is True + assert IndirectManagedNodeAudit.objects.filter(job=bare_job).count() == 1 + audit = IndirectManagedNodeAudit.objects.get(job=bare_job) + assert audit.name == 'vm-foo' + + +@pytest.mark.django_db +def test_save_failure_leaves_flag_false(bare_job, event_query): + """If bulk_create raises, event_queries_processed must remain False for fallback retry.""" + create_registered_event(bare_job) + + with mock.patch.object(IndirectManagedNodeAudit.objects, 'bulk_create', side_effect=RuntimeError('db error')): + save_indirect_host_entries(bare_job.id, wait_for_events=False) + + bare_job.refresh_from_db() + assert bare_job.event_queries_processed is False + + +@pytest.mark.django_db +def test_job_deleted_between_fetch_and_lock(bare_job): + """Race: job exists at initial fetch but is gone by SELECT FOR UPDATE.""" + job_id = bare_job.id + original_atomic = transaction.atomic + + def delete_then_atomic(*args, **kwargs): + Job.objects.filter(id=job_id).delete() + return original_atomic(*args, **kwargs) + + with mock.patch('awx.main.tasks.host_indirect.transaction.atomic', side_effect=delete_then_atomic): + save_indirect_host_entries(job_id, wait_for_events=False)