mirror of
https://github.com/ansible/awx.git
synced 2026-10-29 22:03:20 -02:30
[AAP-88364] Add awx-manage CLI command to enable/disable indirect node counting (#16634)
* feat(cli): add indirect node counting toggle Assisted by gpt-5.6-terra * test(cli): cover indirect node counting toggle command * fix(controller): handle disabled indirect node counting jobs
This commit is contained in:
@@ -34,6 +34,16 @@ register(
|
|||||||
category_slug='system',
|
category_slug='system',
|
||||||
)
|
)
|
||||||
|
|
||||||
|
register(
|
||||||
|
'INDIRECT_NODE_COUNTING_ENABLED',
|
||||||
|
field_class=fields.BooleanField,
|
||||||
|
default=True,
|
||||||
|
label=_('Enable Indirect Node Counting'),
|
||||||
|
help_text=_('Controls whether Controller processes indirect managed nodes reported by jobs.'),
|
||||||
|
category=_('System'),
|
||||||
|
category_slug='system',
|
||||||
|
)
|
||||||
|
|
||||||
register(
|
register(
|
||||||
'ORG_ADMINS_CAN_SEE_ALL_USERS',
|
'ORG_ADMINS_CAN_SEE_ALL_USERS',
|
||||||
field_class=fields.BooleanField,
|
field_class=fields.BooleanField,
|
||||||
|
|||||||
@@ -0,0 +1,21 @@
|
|||||||
|
from django.conf import settings
|
||||||
|
from django.core.management.base import BaseCommand
|
||||||
|
|
||||||
|
from awx.main.tasks.system import clear_setting_cache
|
||||||
|
|
||||||
|
|
||||||
|
class Command(BaseCommand):
|
||||||
|
help = 'Enable or disable indirect node counting.'
|
||||||
|
|
||||||
|
def add_arguments(self, parser):
|
||||||
|
group = parser.add_mutually_exclusive_group(required=True)
|
||||||
|
group.add_argument('--enable', action='store_true', help='Enable indirect node counting.')
|
||||||
|
group.add_argument('--disable', action='store_true', help='Disable indirect node counting.')
|
||||||
|
|
||||||
|
def handle(self, **options):
|
||||||
|
enabled = options['enable']
|
||||||
|
|
||||||
|
settings.INDIRECT_NODE_COUNTING_ENABLED = enabled
|
||||||
|
clear_setting_cache.delay(['INDIRECT_NODE_COUNTING_ENABLED'])
|
||||||
|
state = 'enabled' if enabled else 'disabled'
|
||||||
|
self.stdout.write(f'Indirect node counting is {state}.')
|
||||||
@@ -20,8 +20,6 @@ from awx.main.constants import MINIMAL_EVENTS, ANSIBLE_RUNNER_NEEDS_UPDATE_MESSA
|
|||||||
from awx.main.utils.update_model import update_model
|
from awx.main.utils.update_model import update_model
|
||||||
from awx.main.queue import CallbackQueueDispatcher
|
from awx.main.queue import CallbackQueueDispatcher
|
||||||
|
|
||||||
from flags.state import flag_enabled
|
|
||||||
|
|
||||||
logger = logging.getLogger('awx.main.tasks.callback')
|
logger = logging.getLogger('awx.main.tasks.callback')
|
||||||
|
|
||||||
|
|
||||||
@@ -284,7 +282,7 @@ class RunnerCallback:
|
|||||||
else:
|
else:
|
||||||
logger.warning(f'The file {COLLECTION_FILENAME} unexpectedly did not contain ansible_version')
|
logger.warning(f'The file {COLLECTION_FILENAME} unexpectedly did not contain ansible_version')
|
||||||
|
|
||||||
if flag_enabled("FEATURE_INDIRECT_NODE_COUNTING_ENABLED"):
|
if settings.INDIRECT_NODE_COUNTING_ENABLED:
|
||||||
self.delay_update(event_queries_processed=False)
|
self.delay_update(event_queries_processed=False)
|
||||||
collections_info = collect_queries(query_file_contents)
|
collections_info = collect_queries(query_file_contents)
|
||||||
for collection, data in collections_info.items():
|
for collection, data in collections_info.items():
|
||||||
|
|||||||
@@ -9,9 +9,6 @@ from django.utils.timezone import now, timedelta
|
|||||||
from django.conf import settings
|
from django.conf import settings
|
||||||
from django.db import transaction
|
from django.db import transaction
|
||||||
|
|
||||||
# Django flags
|
|
||||||
from flags.state import flag_enabled
|
|
||||||
|
|
||||||
from dispatcherd.publish import task
|
from dispatcherd.publish import task
|
||||||
from awx.main.dispatch import get_task_queuename
|
from awx.main.dispatch import get_task_queuename
|
||||||
from awx.main.models.indirect_managed_node_audit import IndirectManagedNodeAudit
|
from awx.main.models.indirect_managed_node_audit import IndirectManagedNodeAudit
|
||||||
@@ -166,6 +163,10 @@ def cleanup_old_indirect_host_entries() -> None:
|
|||||||
|
|
||||||
@task(queue=get_task_queuename, timeout=3600 * 5)
|
@task(queue=get_task_queuename, timeout=3600 * 5)
|
||||||
def save_indirect_host_entries(job_id: int, wait_for_events: bool = True) -> None:
|
def save_indirect_host_entries(job_id: int, wait_for_events: bool = True) -> None:
|
||||||
|
if not settings.INDIRECT_NODE_COUNTING_ENABLED:
|
||||||
|
Job.objects.filter(id=job_id, event_queries_processed=False).update(event_queries_processed=True)
|
||||||
|
return
|
||||||
|
|
||||||
try:
|
try:
|
||||||
job = Job.objects.get(id=job_id)
|
job = Job.objects.get(id=job_id)
|
||||||
except Job.DoesNotExist:
|
except Job.DoesNotExist:
|
||||||
@@ -203,7 +204,7 @@ def save_indirect_host_entries(job_id: int, wait_for_events: bool = True) -> Non
|
|||||||
|
|
||||||
@task(queue=get_task_queuename, timeout=3600 * 5)
|
@task(queue=get_task_queuename, timeout=3600 * 5)
|
||||||
def cleanup_and_save_indirect_host_entries_fallback() -> None:
|
def cleanup_and_save_indirect_host_entries_fallback() -> None:
|
||||||
if not flag_enabled("FEATURE_INDIRECT_NODE_COUNTING_ENABLED"):
|
if not settings.INDIRECT_NODE_COUNTING_ENABLED:
|
||||||
return
|
return
|
||||||
|
|
||||||
try:
|
try:
|
||||||
|
|||||||
@@ -428,7 +428,7 @@ class BaseTask(object):
|
|||||||
|
|
||||||
# Copy vendor collections to private_data_dir for indirect node counting
|
# Copy vendor collections to private_data_dir for indirect node counting
|
||||||
# This makes external query files available to the callback plugin in EEs
|
# This makes external query files available to the callback plugin in EEs
|
||||||
if flag_enabled("FEATURE_INDIRECT_NODE_COUNTING_ENABLED"):
|
if settings.INDIRECT_NODE_COUNTING_ENABLED:
|
||||||
vendor_src = '/var/lib/awx/vendor_collections'
|
vendor_src = '/var/lib/awx/vendor_collections'
|
||||||
vendor_dest = os.path.join(private_data_dir, 'vendor_collections')
|
vendor_dest = os.path.join(private_data_dir, 'vendor_collections')
|
||||||
if os.path.exists(vendor_src):
|
if os.path.exists(vendor_src):
|
||||||
@@ -1195,7 +1195,7 @@ class RunJob(SourceControlMixin, BaseTask):
|
|||||||
if 'callbacks_enabled' in config_values:
|
if 'callbacks_enabled' in config_values:
|
||||||
env['ANSIBLE_CALLBACKS_ENABLED'] += ',' + config_values['callbacks_enabled']
|
env['ANSIBLE_CALLBACKS_ENABLED'] += ',' + config_values['callbacks_enabled']
|
||||||
|
|
||||||
if flag_enabled("FEATURE_INDIRECT_NODE_COUNTING_ENABLED"):
|
if settings.INDIRECT_NODE_COUNTING_ENABLED:
|
||||||
env['AWX_COLLECT_HOST_QUERIES'] = '1'
|
env['AWX_COLLECT_HOST_QUERIES'] = '1'
|
||||||
# Add vendor collections path for external query file discovery
|
# Add vendor collections path for external query file discovery
|
||||||
vendor_collections_path = os.path.join(CONTAINER_ROOT, 'vendor_collections')
|
vendor_collections_path = os.path.join(CONTAINER_ROOT, 'vendor_collections')
|
||||||
|
|||||||
@@ -44,8 +44,6 @@ from django.utils.timezone import now, timedelta
|
|||||||
from django.utils.translation import gettext_lazy as _
|
from django.utils.translation import gettext_lazy as _
|
||||||
from django.utils.translation import gettext_noop
|
from django.utils.translation import gettext_noop
|
||||||
|
|
||||||
# Django flags
|
|
||||||
from flags.state import flag_enabled
|
|
||||||
from rest_framework.exceptions import PermissionDenied
|
from rest_framework.exceptions import PermissionDenied
|
||||||
|
|
||||||
# AWX
|
# AWX
|
||||||
@@ -435,7 +433,10 @@ def events_processed_hook(unified_job):
|
|||||||
after the playbook_on_stats/EOF event is processed and final status is saved
|
after the playbook_on_stats/EOF event is processed and final status is saved
|
||||||
Either one of these events could happen before the other, or there may be no events"""
|
Either one of these events could happen before the other, or there may be no events"""
|
||||||
unified_job.send_notification_templates('succeeded' if unified_job.status == 'successful' else 'failed')
|
unified_job.send_notification_templates('succeeded' if unified_job.status == 'successful' else 'failed')
|
||||||
if isinstance(unified_job, Job) and flag_enabled("FEATURE_INDIRECT_NODE_COUNTING_ENABLED"):
|
if isinstance(unified_job, Job):
|
||||||
|
if not settings.INDIRECT_NODE_COUNTING_ENABLED:
|
||||||
|
Job.objects.filter(id=unified_job.id, event_queries_processed=False).update(event_queries_processed=True)
|
||||||
|
return
|
||||||
if unified_job.event_queries_processed is True:
|
if unified_job.event_queries_processed is True:
|
||||||
# If this is called from callback receiver, it likely does not have updated model data
|
# If this is called from callback receiver, it likely does not have updated model data
|
||||||
# a refresh now is formally robust
|
# a refresh now is formally robust
|
||||||
|
|||||||
@@ -0,0 +1,23 @@
|
|||||||
|
import pytest
|
||||||
|
from django.conf import settings
|
||||||
|
from django.core.management import call_command
|
||||||
|
|
||||||
|
from awx.conf.models import Setting
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.django_db
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
('option', 'expected'),
|
||||||
|
[
|
||||||
|
('--enable', True),
|
||||||
|
('--disable', False),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_set_indirect_node_counting_persists_setting(option, expected, mocker):
|
||||||
|
mocker.patch('awx.main.management.commands.set_indirect_node_counting.clear_setting_cache')
|
||||||
|
|
||||||
|
call_command('set_indirect_node_counting', option)
|
||||||
|
|
||||||
|
assert Setting.objects.get(key='INDIRECT_NODE_COUNTING_ENABLED', user=None).value is expected
|
||||||
|
settings._awx_conf_memoizedcache.clear()
|
||||||
|
assert settings.INDIRECT_NODE_COUNTING_ENABLED is expected
|
||||||
@@ -21,7 +21,7 @@ def test_feature_flags_list_endpoint_override(get, flag_val):
|
|||||||
bob = User.objects.create(username='bob', password='test_user', is_superuser=True)
|
bob = User.objects.create(username='bob', password='test_user', is_superuser=True)
|
||||||
|
|
||||||
AAPFlag.objects.all().delete()
|
AAPFlag.objects.all().delete()
|
||||||
flag_name = "FEATURE_INDIRECT_NODE_COUNTING_ENABLED"
|
flag_name = "FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED"
|
||||||
setattr(settings, flag_name, flag_val)
|
setattr(settings, flag_name, flag_val)
|
||||||
seed_feature_flags()
|
seed_feature_flags()
|
||||||
url = "/api/v2/feature_flags/states/"
|
url = "/api/v2/feature_flags/states/"
|
||||||
|
|||||||
@@ -15,6 +15,7 @@ from awx.main.tasks.host_indirect import (
|
|||||||
save_indirect_host_entries,
|
save_indirect_host_entries,
|
||||||
cleanup_and_save_indirect_host_entries_fallback,
|
cleanup_and_save_indirect_host_entries_fallback,
|
||||||
)
|
)
|
||||||
|
from awx.main.tasks.system import events_processed_hook
|
||||||
from awx.main.models.event_query import EventQuery
|
from awx.main.models.event_query import EventQuery
|
||||||
from awx.main.models.indirect_managed_node_audit import IndirectManagedNodeAudit
|
from awx.main.models.indirect_managed_node_audit import IndirectManagedNodeAudit
|
||||||
|
|
||||||
@@ -90,6 +91,32 @@ def create_audit_record(name, job, organization, created=now()):
|
|||||||
return record
|
return record
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.django_db
|
||||||
|
def test_save_indirect_host_entries_skips_when_disabled(bare_job, event_query, settings):
|
||||||
|
create_registered_event(bare_job)
|
||||||
|
settings.INDIRECT_NODE_COUNTING_ENABLED = False
|
||||||
|
|
||||||
|
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 not IndirectManagedNodeAudit.objects.filter(job=bare_job).exists()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.django_db
|
||||||
|
def test_events_processed_hook_discards_pending_counting_when_disabled(bare_job, settings, mocker):
|
||||||
|
settings.INDIRECT_NODE_COUNTING_ENABLED = False
|
||||||
|
mock_delay = mocker.patch('awx.main.tasks.system.save_indirect_host_entries.delay')
|
||||||
|
mock_refresh = mocker.patch.object(bare_job, 'refresh_from_db')
|
||||||
|
mocker.patch('awx.main.models.notifications.JobNotificationMixin.send_notification_templates')
|
||||||
|
|
||||||
|
events_processed_hook(bare_job)
|
||||||
|
|
||||||
|
assert Job.objects.get(id=bare_job.id).event_queries_processed is True
|
||||||
|
mock_refresh.assert_not_called()
|
||||||
|
mock_delay.assert_not_called()
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def event_query():
|
def event_query():
|
||||||
"This is ordinarily created by the artifacts callback"
|
"This is ordinarily created by the artifacts callback"
|
||||||
|
|||||||
@@ -18,7 +18,6 @@ import time
|
|||||||
import yaml
|
import yaml
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from flags.state import enable_flag, disable_flag, flag_enabled
|
|
||||||
|
|
||||||
from awx.main.tests.live.tests.conftest import wait_for_events, unified_job_stdout
|
from awx.main.tests.live.tests.conftest import wait_for_events, unified_job_stdout
|
||||||
from awx.main.tasks.host_indirect import save_indirect_host_entries
|
from awx.main.tasks.host_indirect import save_indirect_host_entries
|
||||||
@@ -49,23 +48,6 @@ VENDOR_COLLECTIONS_BASE = '/var/lib/awx/vendor_collections'
|
|||||||
# --- Fixtures ---
|
# --- Fixtures ---
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def enable_indirect_host_counting():
|
|
||||||
"""Enable FEATURE_INDIRECT_NODE_COUNTING_ENABLED flag for the test.
|
|
||||||
|
|
||||||
Only creates a FlagState DB record if the flag isn't already enabled
|
|
||||||
(e.g. via development_defaults.py), to avoid UniqueViolation errors
|
|
||||||
and to avoid leaking state to other tests.
|
|
||||||
"""
|
|
||||||
flag_name = "FEATURE_INDIRECT_NODE_COUNTING_ENABLED"
|
|
||||||
was_enabled = flag_enabled(flag_name)
|
|
||||||
if not was_enabled:
|
|
||||||
enable_flag(flag_name)
|
|
||||||
yield
|
|
||||||
if not was_enabled:
|
|
||||||
disable_flag(flag_name)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
def vendor_collections_dir():
|
def vendor_collections_dir():
|
||||||
"""Set up mock redhat.indirect_accounting collection at /var/lib/awx/vendor_collections/.
|
"""Set up mock redhat.indirect_accounting collection at /var/lib/awx/vendor_collections/.
|
||||||
@@ -166,7 +148,7 @@ def wait_for_indirect_processing(job, expect_records=True, timeout=5):
|
|||||||
# --- AC8.1: External query populates IndirectManagedNodeAudit correctly ---
|
# --- AC8.1: External query populates IndirectManagedNodeAudit correctly ---
|
||||||
|
|
||||||
|
|
||||||
def test_external_query_populates_audit_table(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
|
def test_external_query_populates_audit_table(live_tmp_folder, run_job_from_playbook, vendor_collections_dir):
|
||||||
"""AC8.1: Job using demo.external.example with external query file populates
|
"""AC8.1: Job using demo.external.example with external query file populates
|
||||||
IndirectManagedNodeAudit table correctly.
|
IndirectManagedNodeAudit table correctly.
|
||||||
|
|
||||||
@@ -198,7 +180,7 @@ def test_external_query_populates_audit_table(live_tmp_folder, run_job_from_play
|
|||||||
# --- AC8.2: Precedence - embedded query takes precedence over external ---
|
# --- AC8.2: Precedence - embedded query takes precedence over external ---
|
||||||
|
|
||||||
|
|
||||||
def test_embedded_query_takes_precedence(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
|
def test_embedded_query_takes_precedence(live_tmp_folder, run_job_from_playbook, vendor_collections_dir):
|
||||||
"""AC8.2: When collection has both embedded and external query files,
|
"""AC8.2: When collection has both embedded and external query files,
|
||||||
the embedded query takes precedence.
|
the embedded query takes precedence.
|
||||||
|
|
||||||
@@ -225,7 +207,7 @@ def test_embedded_query_takes_precedence(live_tmp_folder, run_job_from_playbook,
|
|||||||
# --- AC8.3: Version fallback to compatible version ---
|
# --- AC8.3: Version fallback to compatible version ---
|
||||||
|
|
||||||
|
|
||||||
def test_fallback_to_compatible_version(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
|
def test_fallback_to_compatible_version(live_tmp_folder, run_job_from_playbook, vendor_collections_dir):
|
||||||
"""AC8.3: Job using collection version with no exact query file falls back
|
"""AC8.3: Job using collection version with no exact query file falls back
|
||||||
correctly to compatible version.
|
correctly to compatible version.
|
||||||
|
|
||||||
@@ -257,7 +239,7 @@ def test_fallback_to_compatible_version(live_tmp_folder, run_job_from_playbook,
|
|||||||
# --- AC8.4: Fallback queries don't overcount ---
|
# --- AC8.4: Fallback queries don't overcount ---
|
||||||
|
|
||||||
|
|
||||||
def test_fallback_does_not_overcount(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
|
def test_fallback_does_not_overcount(live_tmp_folder, run_job_from_playbook, vendor_collections_dir):
|
||||||
"""AC8.4: Fallback queries don't count MORE nodes than exact-version queries.
|
"""AC8.4: Fallback queries don't count MORE nodes than exact-version queries.
|
||||||
|
|
||||||
Runs two jobs:
|
Runs two jobs:
|
||||||
@@ -298,7 +280,7 @@ def test_fallback_does_not_overcount(live_tmp_folder, run_job_from_playbook, ena
|
|||||||
# --- AC8.5: Warning logs contain correct version information ---
|
# --- AC8.5: Warning logs contain correct version information ---
|
||||||
|
|
||||||
|
|
||||||
def test_fallback_log_contains_version_info(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
|
def test_fallback_log_contains_version_info(live_tmp_folder, run_job_from_playbook, vendor_collections_dir):
|
||||||
"""AC8.5: Warning logs contain correct version information when fallback is used.
|
"""AC8.5: Warning logs contain correct version information when fallback is used.
|
||||||
|
|
||||||
Runs a job with verbosity=1 so callback plugin verbose output is captured.
|
Runs a job with verbosity=1 so callback plugin verbose output is captured.
|
||||||
@@ -326,7 +308,7 @@ def test_fallback_log_contains_version_info(live_tmp_folder, run_job_from_playbo
|
|||||||
# --- AC8.6: No counting when no compatible fallback exists ---
|
# --- AC8.6: No counting when no compatible fallback exists ---
|
||||||
|
|
||||||
|
|
||||||
def test_no_counting_without_compatible_fallback(live_tmp_folder, run_job_from_playbook, enable_indirect_host_counting, vendor_collections_dir):
|
def test_no_counting_without_compatible_fallback(live_tmp_folder, run_job_from_playbook, vendor_collections_dir):
|
||||||
"""AC8.6: No counting occurs when no compatible fallback exists.
|
"""AC8.6: No counting occurs when no compatible fallback exists.
|
||||||
|
|
||||||
Uses demo.external v3.0.0 with only v1.x external query files available.
|
Uses demo.external v3.0.0 with only v1.x external query files available.
|
||||||
|
|||||||
@@ -1,6 +1,8 @@
|
|||||||
import yaml
|
import yaml
|
||||||
import time
|
import time
|
||||||
|
|
||||||
|
from django.core.management import call_command
|
||||||
|
|
||||||
from awx.main.tests.live.tests.conftest import wait_for_events
|
from awx.main.tests.live.tests.conftest import wait_for_events
|
||||||
from awx.main.tasks.host_indirect import build_indirect_host_data, save_indirect_host_entries
|
from awx.main.tasks.host_indirect import build_indirect_host_data, save_indirect_host_entries
|
||||||
from awx.main.models.indirect_managed_node_audit import IndirectManagedNodeAudit
|
from awx.main.models.indirect_managed_node_audit import IndirectManagedNodeAudit
|
||||||
@@ -64,3 +66,55 @@ def test_indirect_host_counting(live_tmp_folder, run_job_from_playbook):
|
|||||||
assert host_audit.canonical_facts == canonical_facts
|
assert host_audit.canonical_facts == canonical_facts
|
||||||
assert host_audit.facts == facts
|
assert host_audit.facts == facts
|
||||||
assert host_audit.organization == job.organization
|
assert host_audit.organization == job.organization
|
||||||
|
|
||||||
|
|
||||||
|
def test_indirect_host_counting_runtime_toggle(live_tmp_folder, run_job_from_playbook):
|
||||||
|
def run_job(test_name, project=None):
|
||||||
|
result = run_job_from_playbook(
|
||||||
|
test_name,
|
||||||
|
'run_task.yml',
|
||||||
|
scm_url=f'file://{live_tmp_folder}/test_host_query',
|
||||||
|
proj=project,
|
||||||
|
)
|
||||||
|
job = result['job']
|
||||||
|
wait_for_events(job)
|
||||||
|
job.refresh_from_db()
|
||||||
|
if job.event_queries_processed is False:
|
||||||
|
save_indirect_host_entries.delay(job.id, wait_for_events=False)
|
||||||
|
return job, result['project']
|
||||||
|
|
||||||
|
def assert_audit_records(job, expected):
|
||||||
|
if expected:
|
||||||
|
for _ in range(25):
|
||||||
|
if IndirectManagedNodeAudit.objects.filter(job=job).exists():
|
||||||
|
return
|
||||||
|
time.sleep(0.2)
|
||||||
|
assert IndirectManagedNodeAudit.objects.filter(job=job).exists()
|
||||||
|
else:
|
||||||
|
for _ in range(15):
|
||||||
|
assert not IndirectManagedNodeAudit.objects.filter(job=job).exists()
|
||||||
|
time.sleep(0.2)
|
||||||
|
|
||||||
|
def wait_for_setting_propagation():
|
||||||
|
# The dispatcher broadcasts setting cache invalidations to every worker.
|
||||||
|
# Match the wait used by the live-test setting fixture before scheduling
|
||||||
|
# a job that must observe the new value.
|
||||||
|
time.sleep(5)
|
||||||
|
|
||||||
|
call_command('set_indirect_node_counting', '--enable')
|
||||||
|
wait_for_setting_propagation()
|
||||||
|
try:
|
||||||
|
enabled_job, project = run_job('indirect_counting_enabled')
|
||||||
|
assert_audit_records(enabled_job, expected=True)
|
||||||
|
|
||||||
|
call_command('set_indirect_node_counting', '--disable')
|
||||||
|
wait_for_setting_propagation()
|
||||||
|
disabled_job, project = run_job('indirect_counting_disabled', project)
|
||||||
|
assert_audit_records(disabled_job, expected=False)
|
||||||
|
|
||||||
|
call_command('set_indirect_node_counting', '--enable')
|
||||||
|
wait_for_setting_propagation()
|
||||||
|
reenabled_job, _ = run_job('indirect_counting_reenabled', project)
|
||||||
|
assert_audit_records(reenabled_job, expected=True)
|
||||||
|
finally:
|
||||||
|
call_command('set_indirect_node_counting', '--enable')
|
||||||
|
|||||||
@@ -0,0 +1,25 @@
|
|||||||
|
import pytest
|
||||||
|
from django.core.management import call_command
|
||||||
|
from django.core.management.base import CommandError
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.parametrize(
|
||||||
|
('option', 'expected'),
|
||||||
|
[
|
||||||
|
('--enable', True),
|
||||||
|
('--disable', False),
|
||||||
|
],
|
||||||
|
)
|
||||||
|
def test_set_indirect_node_counting(option, expected, mocker):
|
||||||
|
mock_settings = mocker.patch('awx.main.management.commands.set_indirect_node_counting.settings')
|
||||||
|
mock_clear_cache = mocker.patch('awx.main.management.commands.set_indirect_node_counting.clear_setting_cache')
|
||||||
|
|
||||||
|
call_command('set_indirect_node_counting', option)
|
||||||
|
|
||||||
|
assert mock_settings.INDIRECT_NODE_COUNTING_ENABLED is expected
|
||||||
|
mock_clear_cache.delay.assert_called_once_with(['INDIRECT_NODE_COUNTING_ENABLED'])
|
||||||
|
|
||||||
|
|
||||||
|
def test_set_indirect_node_counting_requires_an_option():
|
||||||
|
with pytest.raises(CommandError, match='one of the arguments'):
|
||||||
|
call_command('set_indirect_node_counting')
|
||||||
@@ -67,27 +67,25 @@ SAMPLE_ANSIBLE_DATA = {
|
|||||||
|
|
||||||
|
|
||||||
class TestTryLoadQueryFile:
|
class TestTryLoadQueryFile:
|
||||||
def test_loads_file_without_feature_flag(self):
|
def test_loads_file_when_indirect_node_counting_disabled(self):
|
||||||
with tempfile.TemporaryDirectory() as tmpdir:
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
path = os.path.join(tmpdir, 'ansible_data.json')
|
path = os.path.join(tmpdir, 'ansible_data.json')
|
||||||
with open(path, 'w') as f:
|
with open(path, 'w') as f:
|
||||||
json.dump(SAMPLE_ANSIBLE_DATA, f)
|
json.dump(SAMPLE_ANSIBLE_DATA, f)
|
||||||
|
|
||||||
with mock.patch('awx.main.tasks.callback.flag_enabled', return_value=False):
|
success, data = try_load_query_file(tmpdir)
|
||||||
success, data = try_load_query_file(tmpdir)
|
|
||||||
|
|
||||||
assert success is True
|
assert success is True
|
||||||
assert data['ansible_version'] == '2.16.0'
|
assert data['ansible_version'] == '2.16.0'
|
||||||
assert 'ansible.builtin' in data['installed_collections']
|
assert 'ansible.builtin' in data['installed_collections']
|
||||||
|
|
||||||
def test_loads_file_with_feature_flag(self):
|
def test_loads_file_when_indirect_node_counting_enabled(self):
|
||||||
with tempfile.TemporaryDirectory() as tmpdir:
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
path = os.path.join(tmpdir, 'ansible_data.json')
|
path = os.path.join(tmpdir, 'ansible_data.json')
|
||||||
with open(path, 'w') as f:
|
with open(path, 'w') as f:
|
||||||
json.dump(SAMPLE_ANSIBLE_DATA, f)
|
json.dump(SAMPLE_ANSIBLE_DATA, f)
|
||||||
|
|
||||||
with mock.patch('awx.main.tasks.callback.flag_enabled', return_value=True):
|
success, data = try_load_query_file(tmpdir)
|
||||||
success, data = try_load_query_file(tmpdir)
|
|
||||||
|
|
||||||
assert success is True
|
assert success is True
|
||||||
assert data == SAMPLE_ANSIBLE_DATA
|
assert data == SAMPLE_ANSIBLE_DATA
|
||||||
@@ -101,14 +99,14 @@ class TestTryLoadQueryFile:
|
|||||||
|
|
||||||
|
|
||||||
class TestArtifactsHandler:
|
class TestArtifactsHandler:
|
||||||
def test_always_persists_metadata_when_flag_off(self, mock_me):
|
def test_always_persists_metadata_when_indirect_node_counting_disabled(self, mock_me):
|
||||||
rc = RunnerCallback()
|
rc = RunnerCallback()
|
||||||
with tempfile.TemporaryDirectory() as tmpdir:
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
path = os.path.join(tmpdir, 'ansible_data.json')
|
path = os.path.join(tmpdir, 'ansible_data.json')
|
||||||
with open(path, 'w') as f:
|
with open(path, 'w') as f:
|
||||||
json.dump(SAMPLE_ANSIBLE_DATA, f)
|
json.dump(SAMPLE_ANSIBLE_DATA, f)
|
||||||
|
|
||||||
with mock.patch('awx.main.tasks.callback.flag_enabled', return_value=False):
|
with mock.patch('awx.main.tasks.callback.settings.INDIRECT_NODE_COUNTING_ENABLED', False):
|
||||||
rc.artifacts_handler(tmpdir)
|
rc.artifacts_handler(tmpdir)
|
||||||
|
|
||||||
assert rc.extra_update_fields['installed_collections'] == SAMPLE_ANSIBLE_DATA['installed_collections']
|
assert rc.extra_update_fields['installed_collections'] == SAMPLE_ANSIBLE_DATA['installed_collections']
|
||||||
@@ -117,14 +115,14 @@ class TestArtifactsHandler:
|
|||||||
assert rc.artifacts_processed is True
|
assert rc.artifacts_processed is True
|
||||||
|
|
||||||
@mock.patch('awx.main.tasks.callback.EventQuery')
|
@mock.patch('awx.main.tasks.callback.EventQuery')
|
||||||
def test_creates_event_queries_when_flag_on(self, mock_event_query, mock_me):
|
def test_creates_event_queries_when_indirect_node_counting_enabled(self, mock_event_query, mock_me):
|
||||||
rc = RunnerCallback()
|
rc = RunnerCallback()
|
||||||
with tempfile.TemporaryDirectory() as tmpdir:
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
path = os.path.join(tmpdir, 'ansible_data.json')
|
path = os.path.join(tmpdir, 'ansible_data.json')
|
||||||
with open(path, 'w') as f:
|
with open(path, 'w') as f:
|
||||||
json.dump(SAMPLE_ANSIBLE_DATA, f)
|
json.dump(SAMPLE_ANSIBLE_DATA, f)
|
||||||
|
|
||||||
with mock.patch('awx.main.tasks.callback.flag_enabled', return_value=True):
|
with mock.patch('awx.main.tasks.callback.settings.INDIRECT_NODE_COUNTING_ENABLED', True):
|
||||||
rc.artifacts_handler(tmpdir)
|
rc.artifacts_handler(tmpdir)
|
||||||
|
|
||||||
assert rc.extra_update_fields['installed_collections'] == SAMPLE_ANSIBLE_DATA['installed_collections']
|
assert rc.extra_update_fields['installed_collections'] == SAMPLE_ANSIBLE_DATA['installed_collections']
|
||||||
@@ -133,14 +131,14 @@ class TestArtifactsHandler:
|
|||||||
mock_event_query.assert_called_once()
|
mock_event_query.assert_called_once()
|
||||||
|
|
||||||
@mock.patch('awx.main.tasks.callback.EventQuery')
|
@mock.patch('awx.main.tasks.callback.EventQuery')
|
||||||
def test_no_event_queries_when_flag_off(self, mock_event_query, mock_me):
|
def test_no_event_queries_when_indirect_node_counting_disabled(self, mock_event_query, mock_me):
|
||||||
rc = RunnerCallback()
|
rc = RunnerCallback()
|
||||||
with tempfile.TemporaryDirectory() as tmpdir:
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
path = os.path.join(tmpdir, 'ansible_data.json')
|
path = os.path.join(tmpdir, 'ansible_data.json')
|
||||||
with open(path, 'w') as f:
|
with open(path, 'w') as f:
|
||||||
json.dump(SAMPLE_ANSIBLE_DATA, f)
|
json.dump(SAMPLE_ANSIBLE_DATA, f)
|
||||||
|
|
||||||
with mock.patch('awx.main.tasks.callback.flag_enabled', return_value=False):
|
with mock.patch('awx.main.tasks.callback.settings.INDIRECT_NODE_COUNTING_ENABLED', False):
|
||||||
rc.artifacts_handler(tmpdir)
|
rc.artifacts_handler(tmpdir)
|
||||||
|
|
||||||
mock_event_query.assert_not_called()
|
mock_event_query.assert_not_called()
|
||||||
@@ -148,8 +146,7 @@ class TestArtifactsHandler:
|
|||||||
def test_handles_missing_artifact_file(self, mock_me):
|
def test_handles_missing_artifact_file(self, mock_me):
|
||||||
rc = RunnerCallback()
|
rc = RunnerCallback()
|
||||||
with tempfile.TemporaryDirectory() as tmpdir:
|
with tempfile.TemporaryDirectory() as tmpdir:
|
||||||
with mock.patch('awx.main.tasks.callback.flag_enabled', return_value=False):
|
rc.artifacts_handler(tmpdir)
|
||||||
rc.artifacts_handler(tmpdir)
|
|
||||||
|
|
||||||
assert 'installed_collections' not in rc.extra_update_fields
|
assert 'installed_collections' not in rc.extra_update_fields
|
||||||
assert 'ansible_version' not in rc.extra_update_fields
|
assert 'ansible_version' not in rc.extra_update_fields
|
||||||
|
|||||||
@@ -419,14 +419,13 @@ class TestExternalQueryDiscovery:
|
|||||||
class TestPrivateDataDirIntegration:
|
class TestPrivateDataDirIntegration:
|
||||||
"""Tests for vendor collection copying (AC7.10-AC7.11)."""
|
"""Tests for vendor collection copying (AC7.10-AC7.11)."""
|
||||||
|
|
||||||
@mock.patch('awx.main.tasks.jobs.flag_enabled')
|
@mock.patch('awx.main.tasks.jobs.settings.INDIRECT_NODE_COUNTING_ENABLED', True)
|
||||||
@mock.patch('awx.main.tasks.jobs.shutil.copytree')
|
@mock.patch('awx.main.tasks.jobs.shutil.copytree')
|
||||||
@mock.patch('awx.main.tasks.jobs.os.path.exists')
|
@mock.patch('awx.main.tasks.jobs.os.path.exists')
|
||||||
def test_vendor_collections_copied(self, mock_exists, mock_copytree, mock_flag):
|
def test_vendor_collections_copied(self, mock_exists, mock_copytree):
|
||||||
"""AC7.10: build_private_data_files() copies vendor collections to private_data_dir."""
|
"""AC7.10: build_private_data_files() copies vendor collections to private_data_dir."""
|
||||||
from awx.main.tasks.jobs import BaseTask
|
from awx.main.tasks.jobs import BaseTask
|
||||||
|
|
||||||
mock_flag.return_value = True
|
|
||||||
mock_exists.return_value = True
|
mock_exists.return_value = True
|
||||||
|
|
||||||
task = BaseTask()
|
task = BaseTask()
|
||||||
@@ -439,15 +438,14 @@ class TestPrivateDataDirIntegration:
|
|||||||
|
|
||||||
mock_copytree.assert_called_once_with('/var/lib/awx/vendor_collections', f'{private_data_dir}/vendor_collections')
|
mock_copytree.assert_called_once_with('/var/lib/awx/vendor_collections', f'{private_data_dir}/vendor_collections')
|
||||||
|
|
||||||
@mock.patch('awx.main.tasks.jobs.flag_enabled')
|
@mock.patch('awx.main.tasks.jobs.settings.INDIRECT_NODE_COUNTING_ENABLED', True)
|
||||||
@mock.patch('awx.main.tasks.jobs.logger')
|
@mock.patch('awx.main.tasks.jobs.logger')
|
||||||
@mock.patch('awx.main.tasks.jobs.shutil.copytree')
|
@mock.patch('awx.main.tasks.jobs.shutil.copytree')
|
||||||
@mock.patch('awx.main.tasks.jobs.os.path.exists')
|
@mock.patch('awx.main.tasks.jobs.os.path.exists')
|
||||||
def test_missing_source_handled_gracefully(self, mock_exists, mock_copytree, mock_logger, mock_flag):
|
def test_missing_source_handled_gracefully(self, mock_exists, mock_copytree, mock_logger):
|
||||||
"""AC7.11: Collection copy handles missing source directory gracefully."""
|
"""AC7.11: Collection copy handles missing source directory gracefully."""
|
||||||
from awx.main.tasks.jobs import BaseTask
|
from awx.main.tasks.jobs import BaseTask
|
||||||
|
|
||||||
mock_flag.return_value = True
|
|
||||||
mock_exists.return_value = False
|
mock_exists.return_value = False
|
||||||
|
|
||||||
task = BaseTask()
|
task = BaseTask()
|
||||||
|
|||||||
@@ -8,8 +8,6 @@ LOCAL_SETTINGS = (
|
|||||||
'CACHES',
|
'CACHES',
|
||||||
'DEBUG',
|
'DEBUG',
|
||||||
'NAMED_URL_GRAPH',
|
'NAMED_URL_GRAPH',
|
||||||
# Platform flags are managed by the platform flags system and have environment-specific defaults
|
|
||||||
'FEATURE_INDIRECT_NODE_COUNTING_ENABLED',
|
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -70,24 +68,3 @@ def test_merge_application_name():
|
|||||||
result = merge_application_name(settings)["DATABASES__default__OPTIONS__application_name"]
|
result = merge_application_name(settings)["DATABASES__default__OPTIONS__application_name"]
|
||||||
assert result.startswith("awx-")
|
assert result.startswith("awx-")
|
||||||
assert "test-cluster" in result
|
assert "test-cluster" in result
|
||||||
|
|
||||||
|
|
||||||
def test_development_defaults_feature_flags(monkeypatch):
|
|
||||||
"""Ensure that development_defaults.py sets the correct feature flags."""
|
|
||||||
monkeypatch.setenv('AWX_MODE', 'development')
|
|
||||||
|
|
||||||
# Import the development_defaults module directly to trigger coverage of the new lines
|
|
||||||
import importlib.util
|
|
||||||
import os
|
|
||||||
|
|
||||||
spec = importlib.util.spec_from_file_location("development_defaults", os.path.join(os.path.dirname(__file__), "../../../settings/development_defaults.py"))
|
|
||||||
development_defaults = importlib.util.module_from_spec(spec)
|
|
||||||
spec.loader.exec_module(development_defaults)
|
|
||||||
|
|
||||||
# Also import through the development settings to ensure both paths are tested
|
|
||||||
from awx.settings.development import FEATURE_INDIRECT_NODE_COUNTING_ENABLED
|
|
||||||
|
|
||||||
# Verify the feature flags are set correctly in both the module and settings
|
|
||||||
assert hasattr(development_defaults, 'FEATURE_INDIRECT_NODE_COUNTING_ENABLED')
|
|
||||||
assert development_defaults.FEATURE_INDIRECT_NODE_COUNTING_ENABLED is True
|
|
||||||
assert FEATURE_INDIRECT_NODE_COUNTING_ENABLED is True
|
|
||||||
|
|||||||
@@ -965,7 +965,7 @@ class TestCallbacksEnabled(TestJobExecution):
|
|||||||
|
|
||||||
assert env['ANSIBLE_CALLBACKS_ENABLED'] == 'indirect_instance_count,my_callback'
|
assert env['ANSIBLE_CALLBACKS_ENABLED'] == 'indirect_instance_count,my_callback'
|
||||||
|
|
||||||
def test_collect_host_queries_set_when_flag_on(self, patch_Job, private_data_dir, execution_environment, mock_me):
|
def test_collect_host_queries_set_when_indirect_node_counting_enabled(self, patch_Job, private_data_dir, execution_environment, mock_me):
|
||||||
job = Job(project=Project(), inventory=Inventory())
|
job = Job(project=Project(), inventory=Inventory())
|
||||||
job.execution_environment = execution_environment
|
job.execution_environment = execution_environment
|
||||||
|
|
||||||
@@ -974,12 +974,12 @@ class TestCallbacksEnabled(TestJobExecution):
|
|||||||
task._write_extra_vars_file = mock.Mock()
|
task._write_extra_vars_file = mock.Mock()
|
||||||
|
|
||||||
with mock.patch.object(task, 'build_credentials_list', return_value=[], autospec=True):
|
with mock.patch.object(task, 'build_credentials_list', return_value=[], autospec=True):
|
||||||
with mock.patch('awx.main.tasks.jobs.flag_enabled', return_value=True):
|
with mock.patch('awx.main.tasks.jobs.settings.INDIRECT_NODE_COUNTING_ENABLED', True):
|
||||||
env = task.build_env(job, private_data_dir)
|
env = task.build_env(job, private_data_dir)
|
||||||
|
|
||||||
assert env['AWX_COLLECT_HOST_QUERIES'] == '1'
|
assert env['AWX_COLLECT_HOST_QUERIES'] == '1'
|
||||||
|
|
||||||
def test_collect_host_queries_not_set_when_flag_off(self, patch_Job, private_data_dir, execution_environment, mock_me):
|
def test_collect_host_queries_not_set_when_indirect_node_counting_disabled(self, patch_Job, private_data_dir, execution_environment, mock_me):
|
||||||
job = Job(project=Project(), inventory=Inventory())
|
job = Job(project=Project(), inventory=Inventory())
|
||||||
job.execution_environment = execution_environment
|
job.execution_environment = execution_environment
|
||||||
|
|
||||||
@@ -988,7 +988,8 @@ class TestCallbacksEnabled(TestJobExecution):
|
|||||||
task._write_extra_vars_file = mock.Mock()
|
task._write_extra_vars_file = mock.Mock()
|
||||||
|
|
||||||
with mock.patch.object(task, 'build_credentials_list', return_value=[], autospec=True):
|
with mock.patch.object(task, 'build_credentials_list', return_value=[], autospec=True):
|
||||||
env = task.build_env(job, private_data_dir)
|
with mock.patch('awx.main.tasks.jobs.settings.INDIRECT_NODE_COUNTING_ENABLED', False):
|
||||||
|
env = task.build_env(job, private_data_dir)
|
||||||
|
|
||||||
assert 'AWX_COLLECT_HOST_QUERIES' not in env
|
assert 'AWX_COLLECT_HOST_QUERIES' not in env
|
||||||
|
|
||||||
|
|||||||
@@ -1121,6 +1121,7 @@ SYSTEM_USERNAME = None
|
|||||||
# For indirect host query processing
|
# For indirect host query processing
|
||||||
# if a job is not immediently confirmed to have all events processed
|
# if a job is not immediently confirmed to have all events processed
|
||||||
# it will be eligable for processing after this number of minutes
|
# it will be eligable for processing after this number of minutes
|
||||||
|
INDIRECT_NODE_COUNTING_ENABLED = True
|
||||||
INDIRECT_HOST_QUERY_FALLBACK_MINUTES = 60
|
INDIRECT_HOST_QUERY_FALLBACK_MINUTES = 60
|
||||||
|
|
||||||
# If an error happens in event collection, give up after this time
|
# If an error happens in event collection, give up after this time
|
||||||
@@ -1147,7 +1148,6 @@ OPA_REQUEST_TIMEOUT = 1.5 # The number of seconds after which the connection to
|
|||||||
OPA_REQUEST_RETRIES = 2 # The number of retry attempts for connecting to the OPA server. Default is 2.
|
OPA_REQUEST_RETRIES = 2 # The number of retry attempts for connecting to the OPA server. Default is 2.
|
||||||
|
|
||||||
# feature flags
|
# feature flags
|
||||||
FEATURE_INDIRECT_NODE_COUNTING_ENABLED = False
|
|
||||||
FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED = False
|
FEATURE_OIDC_WORKLOAD_IDENTITY_ENABLED = False
|
||||||
|
|
||||||
# Dispatcher worker lifetime. If set to None, workers will never be retired
|
# Dispatcher worker lifetime. If set to None, workers will never be retired
|
||||||
|
|||||||
@@ -68,4 +68,4 @@ AWX_DISABLE_TASK_MANAGERS = False
|
|||||||
# Needed for launching runserver in debug mode
|
# Needed for launching runserver in debug mode
|
||||||
# ======================!!!!!!! FOR DEVELOPMENT ONLY !!!!!!!=================================
|
# ======================!!!!!!! FOR DEVELOPMENT ONLY !!!!!!!=================================
|
||||||
|
|
||||||
FEATURE_INDIRECT_NODE_COUNTING_ENABLED = True
|
INDIRECT_NODE_COUNTING_ENABLED = True
|
||||||
|
|||||||
@@ -68,6 +68,27 @@ Cluster management
|
|||||||
Do not run other ``awx-manage`` commands unless instructed by Red Hat Ansible personnel.
|
Do not run other ``awx-manage`` commands unless instructed by Red Hat Ansible personnel.
|
||||||
|
|
||||||
|
|
||||||
|
Indirect node counting
|
||||||
|
~~~~~~~~~~~~~~~~~~~~~~
|
||||||
|
|
||||||
|
Red Hat Support can temporarily disable indirect node counting when diagnosing
|
||||||
|
Controller performance issues. Indirect node counting is enabled by default.
|
||||||
|
|
||||||
|
::
|
||||||
|
|
||||||
|
awx-manage set_indirect_node_counting --disable
|
||||||
|
|
||||||
|
To resume processing indirect managed nodes, run:
|
||||||
|
|
||||||
|
::
|
||||||
|
|
||||||
|
awx-manage set_indirect_node_counting --enable
|
||||||
|
|
||||||
|
The command updates the controller setting and clears the setting cache across
|
||||||
|
the cluster. Allow the change to propagate to workers before launching a job
|
||||||
|
that must use the new value. A service restart is not required.
|
||||||
|
|
||||||
|
|
||||||
.. _ag_token_utility:
|
.. _ag_token_utility:
|
||||||
|
|
||||||
Session management
|
Session management
|
||||||
|
|||||||
Reference in New Issue
Block a user