diff --git a/awx/main/analytics/analytics_tasks.py b/awx/main/analytics/analytics_tasks.py index 89c2a27de0..b01610324e 100644 --- a/awx/main/analytics/analytics_tasks.py +++ b/awx/main/analytics/analytics_tasks.py @@ -5,7 +5,7 @@ import logging from dispatcherd.publish import task # AWX -from awx.main.analytics.subsystem_metrics import DispatcherMetrics, CallbackReceiverMetrics +from awx.main.analytics.subsystem_metrics import DispatcherMetrics, CallbackReceiverMetrics, IndirectCountingMetrics from awx.main.dispatch import get_task_queuename logger = logging.getLogger('awx.main.scheduler') @@ -15,3 +15,4 @@ logger = logging.getLogger('awx.main.scheduler') def send_subsystem_metrics(): DispatcherMetrics().send_metrics() CallbackReceiverMetrics().send_metrics() + IndirectCountingMetrics().send_metrics() diff --git a/awx/main/analytics/subsystem_metrics.py b/awx/main/analytics/subsystem_metrics.py index db0295351f..ca20945abb 100644 --- a/awx/main/analytics/subsystem_metrics.py +++ b/awx/main/analytics/subsystem_metrics.py @@ -407,6 +407,20 @@ class DispatcherMetrics(Metrics): super().__init__(settings.METRICS_SERVICE_DISPATCHER, *args, **kwargs) +class IndirectCountingMetrics(Metrics): + """Subsystem metrics for indirect node counting performance and error tracking.""" + + METRICSLIST = [ + IntM('indirect_node_audit_records_created', 'Number of indirect node audit records created'), + FloatM('indirect_node_query_execution_seconds', 'Time spent executing jq queries for indirect node counting'), + IntM('indirect_node_jq_query_errors', 'Number of jq query execution errors during indirect node counting'), + SetFloatM('indirect_node_fallback_cleanup_seconds', 'Time spent in the fallback cleanup job for indirect node counting'), + ] + + def __init__(self, *args, **kwargs): + super().__init__(settings.METRICS_SERVICE_INDIRECT_COUNTING, *args, **kwargs) + + class CallbackReceiverMetrics(Metrics): METRICSLIST = [ SetIntM('callback_receiver_events_queue_size_redis', 'Current number of events in redis queue'), @@ -430,6 +444,7 @@ def metrics(request): output_text = '' output_text += DispatcherMetrics().generate_metrics(request) output_text += CallbackReceiverMetrics().generate_metrics(request) + output_text += IndirectCountingMetrics().generate_metrics(request) dispatcherd_metrics = get_dispatcherd_metrics(request) if dispatcherd_metrics: diff --git a/awx/main/tasks/host_indirect.py b/awx/main/tasks/host_indirect.py index 88e9bb1a93..c42724a61e 100644 --- a/awx/main/tasks/host_indirect.py +++ b/awx/main/tasks/host_indirect.py @@ -1,4 +1,5 @@ import logging +import time from typing import Tuple, Union import yaml @@ -14,6 +15,7 @@ from awx.main.dispatch import get_task_queuename from awx.main.models.indirect_managed_node_audit import IndirectManagedNodeAudit from awx.main.models.event_query import EventQuery from awx.main.models import Job +from awx.main.analytics.subsystem_metrics import IndirectCountingMetrics logger = logging.getLogger(__name__) @@ -36,93 +38,114 @@ def get_hashable_form(input_data: Union[dict, list, Tuple, int, float, str, bool raise UnhashableFacts(f'Cannonical facts contains a {type(input_data)} type which can not be hashed.') -def build_indirect_host_data(job: Job, job_event_queries: dict[str, dict[str, str]]) -> list[IndirectManagedNodeAudit]: - results = {} - compiled_jq_expressions = {} # Cache for compiled jq expressions - facts_missing_logged = False - unhashable_facts_logged = False - name_missing_logged = set() +def _execute_jq_query(compiled_jq, event_data_res, resolved_action, event_id, s_metrics): + """Execute a compiled jq expression against event data, recording metrics.""" + jq_start = time.perf_counter() + try: + return compiled_jq.input(event_data_res).all() + except Exception as e: + s_metrics.inc('indirect_node_jq_query_errors', 1) + logger.warning(f'jq error for module {resolved_action} on event {event_id}: {e}') + return None + finally: + s_metrics.inc('indirect_node_query_execution_seconds', time.perf_counter() - jq_start) - job_event_queries_fqcn = {} + +def _build_wildcard_queries(job_event_queries: dict[str, dict[str, str]]) -> dict[str, dict[str, str]]: + """Build a lookup of wildcard (a.b.*) queries keyed by fqcn (a.b).""" + fqcn_queries = {} for query_k, query_v in job_event_queries.items(): if len(parts := query_k.split('.')) != 3: logger.info(f"Skiping malformed query '{query_k}'. Expected to be of the form 'a.b.c'") continue - if parts[2] != '*': - continue - job_event_queries_fqcn['.'.join(parts[0:2])] = query_v + if parts[2] == '*': + fqcn_queries['.'.join(parts[0:2])] = query_v + return fqcn_queries + + +def _resolve_event_query(event, job_event_queries, job_event_queries_fqcn): + """Match an event to its jq query string. Returns (resolved_action, jq_str) or (None, None).""" + if 'res' not in event.event_data: + return None, None + + resolved_action = event.event_data.get('resolved_action', None) + if not resolved_action: + return None, None + + if len(resolved_action_parts := resolved_action.split('.')) != 3: + logger.debug(f"Malformed invocation module name '{resolved_action}'. Expected to be of the form 'a.b.c'") + return None, None + + resolved_action_fqcn = '.'.join(resolved_action_parts[0:2]) + jq_str = job_event_queries.get(resolved_action, job_event_queries_fqcn.get(resolved_action_fqcn, {})).get('query') + if not jq_str: + return None, None + + return resolved_action, jq_str + + +def _process_jq_result(data, resolved_action, event, jq_str, job, results, log_state): + """Process a single jq result into an audit record.""" + if not data.get('canonical_facts'): + if not log_state['facts_missing']: + logger.error(f'jq output missing canonical_facts for module {resolved_action} on event {event.id} using jq:{jq_str}') + log_state['facts_missing'] = True + return + + canonical_facts = data['canonical_facts'] + try: + hashable_facts = get_hashable_form(canonical_facts) + except UnhashableFacts: + if not log_state['unhashable']: + logger.info(f'Could not hash canonical_facts {canonical_facts}, skipping') + log_state['unhashable'] = True + return + + name = data.get('name') + if name is None: + if resolved_action not in log_state['name_missing']: + logger.warning(f'jq output missing name for module {resolved_action} on event {event.id} using jq:{jq_str}') + log_state['name_missing'].add(resolved_action) + return + + if hashable_facts in results: + audit_record = results[hashable_facts] + else: + audit_record = IndirectManagedNodeAudit( + canonical_facts=canonical_facts, + facts=data.get('facts'), + job=job, + organization=job.organization, + name=name, + ) + results[hashable_facts] = audit_record + + if resolved_action not in audit_record.events: + audit_record.events.append(resolved_action) + audit_record.count += 1 + + +def build_indirect_host_data(job: Job, job_event_queries: dict[str, dict[str, str]], s_metrics: IndirectCountingMetrics) -> list[IndirectManagedNodeAudit]: + results = {} + compiled_jq_expressions = {} + job_event_queries_fqcn = _build_wildcard_queries(job_event_queries) + log_state = {'facts_missing': False, 'unhashable': False, 'name_missing': set()} for event in job.job_events.filter(event_data__isnull=False).iterator(): - if 'res' not in event.event_data: + resolved_action, jq_str = _resolve_event_query(event, job_event_queries, job_event_queries_fqcn) + if not resolved_action: continue - if not (resolved_action := event.event_data.get('resolved_action', None)): - continue + if jq_str not in compiled_jq_expressions: + compiled_jq_expressions[jq_str] = jq.compile(jq_str) + compiled_jq = compiled_jq_expressions[jq_str] - if len(resolved_action_parts := resolved_action.split('.')) != 3: - logger.debug(f"Malformed invocation module name '{resolved_action}'. Expected to be of the form 'a.b.c'") - continue - - resolved_action_fqcn = '.'.join(resolved_action_parts[0:2]) - - # Match module invocation to collection queries - # First match against fully qualified query names i.e. a.b.c - # Then try and match against wildcard queries i.e. a.b.* - if not (jq_str_for_event := job_event_queries.get(resolved_action, job_event_queries_fqcn.get(resolved_action_fqcn, {})).get('query')): - continue - - # Recall from cache, or process the jq expression, and loop over the jq results - if jq_str_for_event not in compiled_jq_expressions: - compiled_jq_expressions[resolved_action] = jq.compile(jq_str_for_event) - compiled_jq = compiled_jq_expressions[resolved_action] - - try: - data_source = compiled_jq.input(event.event_data['res']).all() - except Exception as e: - logger.warning(f'error for module {resolved_action} and data {event.event_data["res"]}: {e}') + data_source = _execute_jq_query(compiled_jq, event.event_data['res'], resolved_action, event.id, s_metrics) + if data_source is None: continue for data in data_source: - # From this jq result (specific to a single Ansible module), get index information about this host record - if not data.get('canonical_facts'): - if not facts_missing_logged: - logger.error(f'jq output missing canonical_facts for module {resolved_action} on event {event.id} using jq:{jq_str_for_event}') - facts_missing_logged = True - continue - canonical_facts = data['canonical_facts'] - try: - hashable_facts = get_hashable_form(canonical_facts) - except UnhashableFacts: - if not unhashable_facts_logged: - logger.info(f'Could not hash canonical_facts {canonical_facts}, skipping') - unhashable_facts_logged = True - continue - - # 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] - else: - audit_record = IndirectManagedNodeAudit( - canonical_facts=canonical_facts, - facts=facts, - job=job, - organization=job.organization, - name=name, - ) - results[hashable_facts] = audit_record - - # Increment rolling count fields - if resolved_action not in audit_record.events: - audit_record.events.append(resolved_action) - audit_record.count += 1 + _process_jq_result(data, resolved_action, event, jq_str, job, results, log_state) return list(results.values()) @@ -145,11 +168,13 @@ def fetch_job_event_query(job: Job) -> dict[str, dict[str, str]]: return net_job_data -def save_indirect_host_entries_of_job(job: Job) -> None: +def save_indirect_host_entries_of_job(job: Job, s_metrics: IndirectCountingMetrics) -> None: "Once we have a job and we know that we want to do indirect host processing, this is called" 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, s_metrics) IndirectManagedNodeAudit.objects.bulk_create(records) + record_count = len(records) + transaction.on_commit(lambda: s_metrics.inc('indirect_node_audit_records_created', record_count)) def cleanup_old_indirect_host_entries() -> None: @@ -181,6 +206,7 @@ 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') + s_metrics = IndirectCountingMetrics() try: with transaction.atomic(): try: @@ -195,11 +221,16 @@ def save_indirect_host_entries(job_id: int, wait_for_events: bool = True) -> Non # 2. the artifacts_handler has not yet been called for this job return - save_indirect_host_entries_of_job(job) + save_indirect_host_entries_of_job(job, s_metrics) 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}') + finally: + try: + s_metrics.pipe_execute() + except Exception: + logger.warning("Failed to flush indirect counting metrics") @task(queue=get_task_queuename, timeout=3600 * 5) @@ -207,6 +238,9 @@ def cleanup_and_save_indirect_host_entries_fallback() -> None: if not settings.INDIRECT_NODE_COUNTING_ENABLED: return + s_metrics = IndirectCountingMetrics() + fallback_start = time.perf_counter() + try: cleanup_old_indirect_host_entries() except Exception as e: @@ -221,3 +255,9 @@ def cleanup_and_save_indirect_host_entries_fallback() -> None: job_ct += 1 if job_ct: logger.info(f'Restarted event processing for {job_ct} jobs') + + s_metrics.set('indirect_node_fallback_cleanup_seconds', time.perf_counter() - fallback_start) + try: + s_metrics.pipe_execute() + except Exception: + logger.warning("Failed to flush indirect counting metrics") diff --git a/awx/main/tests/functional/analytics/test_metrics.py b/awx/main/tests/functional/analytics/test_metrics.py index 2652db85b7..55082b45e2 100644 --- a/awx/main/tests/functional/analytics/test_metrics.py +++ b/awx/main/tests/functional/analytics/test_metrics.py @@ -1,3 +1,5 @@ +from unittest import mock + import pytest from django.test import RequestFactory @@ -5,6 +7,9 @@ from prometheus_client.parser import text_string_to_metric_families from rest_framework.request import Request from awx.main import models from awx.main.analytics.metrics import metrics +from awx.main.analytics.analytics_tasks import send_subsystem_metrics +import awx.main.analytics.subsystem_metrics as subsystem_metrics_module +from awx.main.analytics.subsystem_metrics import IndirectCountingMetrics from awx.main.analytics.dispatcherd_metrics import get_dispatcherd_metrics from awx.api.versioning import reverse @@ -132,3 +137,37 @@ def test_dispatcherd_metrics_metric_filter_excludes_unrelated(mocker): request = Request(RequestFactory().get('/api/v2/metrics/', {'metric': 'awx_system_info'})) assert get_dispatcherd_metrics(request) == '' + + +@mock.patch('awx.main.analytics.analytics_tasks.IndirectCountingMetrics') +@mock.patch('awx.main.analytics.analytics_tasks.CallbackReceiverMetrics') +@mock.patch('awx.main.analytics.analytics_tasks.DispatcherMetrics') +def test_send_subsystem_metrics(mock_dispatcher, mock_callback, mock_indirect): + """send_subsystem_metrics calls send_metrics on all subsystem metric classes.""" + send_subsystem_metrics() + mock_dispatcher.return_value.send_metrics.assert_called_once() + mock_callback.return_value.send_metrics.assert_called_once() + mock_indirect.return_value.send_metrics.assert_called_once() + + +@pytest.mark.django_db +def test_indirect_counting_metrics_init(): + """IndirectCountingMetrics can be instantiated and has expected metric fields.""" + m = IndirectCountingMetrics(metrics_have_changed=False) + assert 'indirect_node_audit_records_created' in m.METRICS + assert 'indirect_node_query_execution_seconds' in m.METRICS + assert 'indirect_node_jq_query_errors' in m.METRICS + assert 'indirect_node_fallback_cleanup_seconds' in m.METRICS + + +@mock.patch.object(subsystem_metrics_module, 'IndirectCountingMetrics') +@mock.patch.object(subsystem_metrics_module, 'CallbackReceiverMetrics') +@mock.patch.object(subsystem_metrics_module, 'DispatcherMetrics') +@mock.patch.object(subsystem_metrics_module, 'get_dispatcherd_metrics', return_value='') +def test_subsystem_metrics_function_includes_indirect_counting(mock_dispatcherd, mock_dispatcher, mock_callback, mock_indirect): + """subsystem_metrics.metrics() calls generate_metrics on IndirectCountingMetrics.""" + for m in (mock_dispatcher, mock_callback, mock_indirect): + m.return_value.generate_metrics.return_value = '' + request = mock.MagicMock() + subsystem_metrics_module.metrics(request) + mock_indirect.return_value.generate_metrics.assert_called_once_with(request) diff --git a/awx/main/tests/functional/tasks/test_host_indirect.py b/awx/main/tests/functional/tasks/test_host_indirect.py index 549b2c07fe..2f3fa8d07a 100644 --- a/awx/main/tests/functional/tasks/test_host_indirect.py +++ b/awx/main/tests/functional/tasks/test_host_indirect.py @@ -10,6 +10,9 @@ from django.utils.timezone import now, timedelta from awx.main.models import Job from awx.main.tasks.host_indirect import ( + _execute_jq_query, + _resolve_event_query, + _process_jq_result, build_indirect_host_data, fetch_job_event_query, save_indirect_host_entries, @@ -183,7 +186,7 @@ def new_audit_record(bare_job, organization): ), ) def test_build_indirect_host_data(job_with_counted_event, queries: Query, expected_matches: int): - data = build_indirect_host_data(job_with_counted_event, {k: v for d in queries for k, v in d.items()}) + data = build_indirect_host_data(job_with_counted_event, {k: v for d in queries for k, v in d.items()}, mock.MagicMock()) assert len(data) == expected_matches @@ -208,7 +211,7 @@ def test_build_indirect_host_data(job_with_counted_event, queries: Query, expect ) def test_build_indirect_host_data_malformed_module_name(mock_logger_debug, bare_job, task_name: str): create_registered_event(bare_job, task_name) - assert build_indirect_host_data(bare_job, Query('demo.query.example', TEST_JQ)) == [] + assert build_indirect_host_data(bare_job, Query('demo.query.example', TEST_JQ), mock.MagicMock()) == [] mock_logger_debug.assert_called_once_with(f"Malformed invocation module name '{task_name}'. Expected to be of the form 'a.b.c'") @@ -232,7 +235,7 @@ def test_build_indirect_host_data_malformed_module_name(mock_logger_debug, bare_ ), ) def test_build_indirect_host_data_malformed_query(mock_logger_info, job_with_counted_event, query: str): - assert build_indirect_host_data(job_with_counted_event, {query: {'query': TEST_JQ}}) == [] + assert build_indirect_host_data(job_with_counted_event, {query: {'query': TEST_JQ}}, mock.MagicMock()) == [] mock_logger_info.assert_called_once_with(f"Skiping malformed query '{query}'. Expected to be of the form 'a.b.c'") @@ -407,7 +410,7 @@ def test_null_name_event_skipped_with_warning(mock_logger_warning, bare_job, eve } ) query = Query('demo.query.example', TEST_JQ) - records = build_indirect_host_data(bare_job, query) + records = build_indirect_host_data(bare_job, query, mock.MagicMock()) assert records == [] mock_logger_warning.assert_called_once() assert 'missing name' in mock_logger_warning.call_args[0][0] @@ -449,6 +452,120 @@ def test_save_failure_leaves_flag_false(bare_job, event_query): assert bare_job.event_queries_processed is False +@pytest.mark.django_db +def test_execute_jq_query_success(): + """Successful jq execution returns results and records timing.""" + compiled_jq = mock.MagicMock() + compiled_jq.input.return_value.all.return_value = [{'name': 'vm-1'}] + s_metrics = mock.MagicMock() + + result = _execute_jq_query(compiled_jq, {'host': 'foo'}, 'demo.query.example', 42, s_metrics) + + assert result == [{'name': 'vm-1'}] + compiled_jq.input.assert_called_once_with({'host': 'foo'}) + s_metrics.inc.assert_called_once_with('indirect_node_query_execution_seconds', mock.ANY) + + +@pytest.mark.django_db +def test_execute_jq_query_error(): + """Failed jq execution returns None, increments error counter, and records timing.""" + compiled_jq = mock.MagicMock() + compiled_jq.input.return_value.all.side_effect = ValueError('bad input') + s_metrics = mock.MagicMock() + + result = _execute_jq_query(compiled_jq, {'host': 'foo'}, 'demo.query.example', 42, s_metrics) + + assert result is None + s_metrics.inc.assert_any_call('indirect_node_jq_query_errors', 1) + s_metrics.inc.assert_any_call('indirect_node_query_execution_seconds', mock.ANY) + + +@pytest.mark.django_db +def test_jq_runtime_error_increments_error_metric(bare_job, event_query): + """A jq query that errors at runtime should increment the error metric.""" + create_registered_event(bare_job) + s_metrics = mock.MagicMock() + query = Query('demo.query.example', TEST_JQ) + mock_compiled = mock.MagicMock() + mock_compiled.input.return_value.all.side_effect = ValueError('jq runtime error') + with mock.patch('jq.compile', return_value=mock_compiled): + records = build_indirect_host_data(bare_job, query, s_metrics) + assert records == [] + s_metrics.inc.assert_any_call('indirect_node_jq_query_errors', 1) + s_metrics.inc.assert_any_call('indirect_node_query_execution_seconds', mock.ANY) + + +@pytest.mark.django_db +def test_resolve_event_query_missing_res(): + """Events without 'res' in event_data return (None, None).""" + event = mock.MagicMock() + event.event_data = {'resolved_action': 'demo.query.example'} + assert _resolve_event_query(event, {}, {}) == (None, None) + + +@pytest.mark.django_db +def test_resolve_event_query_missing_resolved_action(): + """Events without 'resolved_action' return (None, None).""" + event = mock.MagicMock() + event.event_data = {'res': {}} + assert _resolve_event_query(event, {}, {}) == (None, None) + + +@pytest.mark.django_db +def test_process_jq_result_unhashable_facts(): + """Unhashable canonical_facts should be skipped.""" + data = {'canonical_facts': {frozenset(): 'bad'}, 'name': 'vm-1'} + log_state = {'facts_missing': False, 'unhashable': False, 'name_missing': set()} + results = {} + event = mock.MagicMock() + _process_jq_result(data, 'demo.query.example', event, 'jq_str', mock.MagicMock(), results, log_state) + assert results == {} + assert log_state['unhashable'] is True + + +@mock.patch('awx.main.tasks.host_indirect.IndirectCountingMetrics') +@pytest.mark.django_db +def test_pipe_execute_failure_logs_warning_save(mock_metrics_cls, bare_job, event_query): + """Defensive: pipe_execute failure in save_indirect_host_entries logs warning but doesn't raise.""" + create_registered_event(bare_job) + mock_instance = mock.MagicMock() + mock_instance.pipe_execute.side_effect = RuntimeError('redis down') + mock_metrics_cls.return_value = mock_instance + + save_indirect_host_entries(bare_job.id, wait_for_events=False) + + bare_job.refresh_from_db() + assert bare_job.event_queries_processed is True + + +@mock.patch('awx.main.tasks.host_indirect.IndirectCountingMetrics') +@pytest.mark.django_db +def test_pipe_execute_failure_logs_warning_fallback(mock_metrics_cls, old_audit_record, new_audit_record): + """Defensive: pipe_execute failure in fallback logs warning but doesn't raise.""" + mock_instance = mock.MagicMock() + mock_instance.pipe_execute.side_effect = RuntimeError('redis down') + mock_metrics_cls.return_value = mock_instance + + cleanup_and_save_indirect_host_entries_fallback() + + assert IndirectManagedNodeAudit.objects.count() < 2 + + +@pytest.mark.django_db +def test_on_commit_increments_audit_records_metric(bare_job, event_query): + """The audit record count metric should fire via on_commit.""" + create_registered_event(bare_job) + s_metrics = mock.MagicMock() + callbacks = [] + with mock.patch('awx.main.tasks.host_indirect.transaction.on_commit', side_effect=lambda fn: callbacks.append(fn)): + from awx.main.tasks.host_indirect import save_indirect_host_entries_of_job + + save_indirect_host_entries_of_job(bare_job, s_metrics) + assert len(callbacks) == 1 + callbacks[0]() + s_metrics.inc.assert_any_call('indirect_node_audit_records_created', mock.ANY) + + @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.""" @@ -461,3 +578,45 @@ def test_job_deleted_between_fetch_and_lock(bare_job): 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) + + +def test_process_jq_result_missing_canonical_facts(): + """Data without canonical_facts should be skipped and an error logged once; second call skips logging.""" + data = {'name': 'vm-1'} + log_state = {'facts_missing': False, 'unhashable': False, 'name_missing': set()} + results = {} + event = mock.MagicMock() + with mock.patch('awx.main.tasks.host_indirect.logger') as mock_logger: + _process_jq_result(data, 'demo.query.example', event, 'jq_str', mock.MagicMock(), results, log_state) + assert results == {} + assert log_state['facts_missing'] is True + mock_logger.error.assert_called_once() + + with mock.patch('awx.main.tasks.host_indirect.logger') as mock_logger2: + _process_jq_result(data, 'demo.query.example', event, 'jq_str', mock.MagicMock(), results, log_state) + assert results == {} + mock_logger2.error.assert_not_called() + + +def test_process_jq_result_unhashable_already_logged(): + """Second unhashable fact does not log again when already flagged.""" + data = {'canonical_facts': {frozenset(): 'bad'}, 'name': 'vm-1'} + log_state = {'facts_missing': False, 'unhashable': True, 'name_missing': set()} + results = {} + event = mock.MagicMock() + with mock.patch('awx.main.tasks.host_indirect.logger') as mock_logger: + _process_jq_result(data, 'demo.query.example', event, 'jq_str', mock.MagicMock(), results, log_state) + assert results == {} + mock_logger.info.assert_not_called() + + +def test_process_jq_result_null_name_already_logged(): + """Second null-name event for the same resolved_action does not log again.""" + data = {'canonical_facts': {'host_name': 'foo'}, 'name': None} + log_state = {'facts_missing': False, 'unhashable': False, 'name_missing': {'demo.query.example'}} + results = {} + event = mock.MagicMock() + with mock.patch('awx.main.tasks.host_indirect.logger') as mock_logger: + _process_jq_result(data, 'demo.query.example', event, 'jq_str', mock.MagicMock(), results, log_state) + assert results == {} + mock_logger.warning.assert_not_called() diff --git a/awx/main/tests/live/tests/test_indirect_host_counting.py b/awx/main/tests/live/tests/test_indirect_host_counting.py index 75f3666ba6..7e0a8f14b6 100644 --- a/awx/main/tests/live/tests/test_indirect_host_counting.py +++ b/awx/main/tests/live/tests/test_indirect_host_counting.py @@ -1,5 +1,6 @@ import yaml import time +from unittest import mock from django.core.management import call_command @@ -20,7 +21,7 @@ def test_indirect_host_counting(live_tmp_folder, run_job_from_playbook): event_query = {'demo.query.example': {'query': module_jq_str}} # Run the task logic directly with local data - results = build_indirect_host_data(job, event_query) + results = build_indirect_host_data(job, event_query, mock.MagicMock()) assert len(results) == 1 host_audit_entry = results[0] diff --git a/awx/settings/defaults.py b/awx/settings/defaults.py index bc43a5504a..8143e306df 100644 --- a/awx/settings/defaults.py +++ b/awx/settings/defaults.py @@ -1007,6 +1007,7 @@ HOST_METRIC_SUMMARY_TASK_INTERVAL = 7 # days METRICS_SERVICE_CALLBACK_RECEIVER = 'callback_receiver' METRICS_SERVICE_DISPATCHER = 'dispatcherd' METRICS_SERVICE_WEBSOCKETS = 'websockets' +METRICS_SERVICE_INDIRECT_COUNTING = 'indirect_counting' METRICS_SUBSYSTEM_CONFIG = { 'server': {