AAP-91428 | feat: subsystem metrics for indirect node counting (#16649)

* AAP-91428 | feat: subsystem metrics for indirect node counting

  Expose Prometheus metrics for indirect node counting at
  /api/v2/metrics?subsystemonly=1, following the existing DispatcherMetrics
  pattern. This is a prerequisite for perf/scale testing (AAP-57413) before
  the feature flag can be removed for GA in 2.7.

  Metrics added:
  - indirect_node_audit_records_created (counter)
  - indirect_node_query_execution_seconds (accumulated float)
  - indirect_node_jq_query_errors (counter)
  - indirect_node_fallback_cleanup_seconds (latest-value float)

  Instrumentation points in host_indirect.py:
  - jq query execution timing and error counting in build_indirect_host_data
  - audit record count after bulk_create in save_indirect_host_entries_of_job
  - fallback cleanup job wall-clock timing

  Also registers IndirectCountingMetrics in send_subsystem_metrics for
  cross-node metric broadcasting.

Co-Authored-By: Claude Code <noreply@anthropic.com>

* AAP-91428 | fix: SonarQube complaints (coverage + complexity)

* AAP-91428 | fix: use jq_str and fix docstring

* AAP-91428 | test: tests to get coverage up

Co-Authored-By: Claude Code <noreply@anthropic.com>

---------

Co-authored-by: Claude Code <noreply@anthropic.com>

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()

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:

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")

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)

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()

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]
 

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': {