mirror of
https://github.com/ansible/awx.git
synced 2026-10-31 06:26:15 -02:30
[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) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.6
parent
3917f3bf74
commit
eb7090dd0f
@@ -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
|
compiled_jq_expressions = {} # Cache for compiled jq expressions
|
||||||
facts_missing_logged = False
|
facts_missing_logged = False
|
||||||
unhashable_facts_logged = False
|
unhashable_facts_logged = False
|
||||||
|
name_missing_logged = set()
|
||||||
|
|
||||||
job_event_queries_fqcn = {}
|
job_event_queries_fqcn = {}
|
||||||
for query_k, query_v in job_event_queries.items():
|
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
|
# Obtain the record based on the hashable canonical_facts now determined
|
||||||
facts = data.get('facts')
|
facts = data.get('facts')
|
||||||
name = data.get('name')
|
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:
|
if hashable_facts in results:
|
||||||
audit_record = results[hashable_facts]
|
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)
|
job_event_queries = fetch_job_event_query(job)
|
||||||
records = build_indirect_host_data(job, job_event_queries)
|
records = build_indirect_host_data(job, job_event_queries)
|
||||||
IndirectManagedNodeAudit.objects.bulk_create(records)
|
IndirectManagedNodeAudit.objects.bulk_create(records)
|
||||||
job.event_queries_processed = True
|
|
||||||
|
|
||||||
|
|
||||||
def cleanup_old_indirect_host_entries() -> None:
|
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
|
return
|
||||||
job.log_lifecycle(f'finished processing {current_events} events, running save_indirect_host_entries')
|
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:
|
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:
|
except Exception:
|
||||||
logger.exception(f'Error processing indirect host data for job_id={job_id}')
|
logger.exception(f'Error processing indirect host data for job_id={job_id}')
|
||||||
|
|
||||||
|
|||||||
@@ -4,8 +4,11 @@ from unittest import mock
|
|||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
from django.db import transaction
|
||||||
from django.utils.timezone import now, timedelta
|
from django.utils.timezone import now, timedelta
|
||||||
|
|
||||||
|
from awx.main.models import Job
|
||||||
|
|
||||||
from awx.main.tasks.host_indirect import (
|
from awx.main.tasks.host_indirect import (
|
||||||
build_indirect_host_data,
|
build_indirect_host_data,
|
||||||
fetch_job_event_query,
|
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()
|
cleanup_and_save_indirect_host_entries_fallback()
|
||||||
count_after_cleanup = IndirectManagedNodeAudit.objects.count()
|
count_after_cleanup = IndirectManagedNodeAudit.objects.count()
|
||||||
assert count_after_cleanup == 1
|
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)
|
||||||
|
|||||||
Reference in New Issue
Block a user