diff --git a/awx/main/dispatch/reaper.py b/awx/main/dispatch/reaper.py index 70ec0812a5..299dab768f 100644 --- a/awx/main/dispatch/reaper.py +++ b/awx/main/dispatch/reaper.py @@ -8,24 +8,6 @@ from awx.main.models import Instance, UnifiedJob, WorkflowJob logger = logging.getLogger('awx.main.dispatch') -def startup_reaping(): - """ - If this particular instance is starting, then we know that any running jobs are invalid - so we will reap those jobs as a special action here - """ - jobs = UnifiedJob.objects.filter(status='running', controller_node=Instance.objects.my_hostname()) - job_ids = [] - for j in jobs: - job_ids.append(j.id) - reap_job( - j, - 'failed', - job_explanation='Task was marked as running at system start up. The system must have not shut down properly, so it has been marked as failed.', - ) - if job_ids: - logger.error(f'Unified jobs {job_ids} were reaped on dispatch startup') - - def reap_job(j, status, job_explanation=None): j.refresh_from_db(fields=['status', 'job_explanation']) status_before = j.status @@ -48,8 +30,11 @@ def reap_job(j, status, job_explanation=None): def reap(instance=None, status='failed', job_explanation=None, excluded_uuids=None, ref_time=None): - """ - Reap all jobs in running for this instance. + """Reap all running jobs for this instance. + + Per-job adopt-or-reap decisions (based on work_unit_id) belong in the caller's loop, + not here as a queryset filter. See _process_startup_jobs and _process_running_jobs + in awx.main.tasks.system for the unified loops that replaced undispatched_only filtering. """ if instance is None: hostname = Instance.objects.my_hostname() diff --git a/awx/main/dispatch/worker/callback.py b/awx/main/dispatch/worker/callback.py index 5673bbb9b9..294f3ab1ef 100644 --- a/awx/main/dispatch/worker/callback.py +++ b/awx/main/dispatch/worker/callback.py @@ -195,7 +195,11 @@ class CallbackBrokerWorker: def flush(self, force=False): now = tz_now() - if force or (time.time() - self.last_flush) > settings.JOB_EVENT_BUFFER_SECONDS or any([len(events) >= 1000 for events in self.buff.values()]): + if ( + force + or (time.time() - self.last_flush) > settings.JOB_EVENT_BUFFER_SECONDS + or any(len(events) >= settings.JOB_EVENT_CALLBACK_BUFFER_SIZE for events in self.buff.values()) + ): metrics_bulk_events_saved = 0 metrics_singular_events_saved = 0 metrics_events_batch_save_errors = 0 diff --git a/awx/main/tasks/callback.py b/awx/main/tasks/callback.py index 6e86718a5f..fa21c31f3c 100644 --- a/awx/main/tasks/callback.py +++ b/awx/main/tasks/callback.py @@ -89,6 +89,8 @@ class RunnerCallback: self.wrapup_event_dispatched = False self.artifacts_processed = False self.extra_update_fields = {} + self.dedup_threshold = None # skip events with counter <= this (safe contiguous range already in DB) + self.persisted_counters = None # skip events in this set (collision zone: counters above threshold already in DB) def update_model(self, pk, _attempt=0, **updates): return update_model(self.model, pk, _attempt=0, _max_attempts=self.update_attempts, **updates) @@ -121,6 +123,57 @@ class RunnerCallback: self.delay_update(result_traceback=ANSIBLE_RUNNER_NEEDS_UPDATE_MESSAGE) return self.extra_update_fields + def configure_for_job(self, instance, safe_env=None, dedup_threshold=None, persisted_counters=None): + """Initialize callback fields for event processing. + + This is the single source of truth for RunnerCallback initialization, shared + between the normal job path (BaseTask.run) and adoption (reattach_to_work_unit). + Having both paths call this method means any missing field is caught by the + normal test suite rather than only by adoption-specific tests. + + Args: + instance: the UnifiedJob model instance being run + safe_env: masked environment dict for log output. None → {} (adoption path, + where credentials are not available at reconnect time) + dedup_threshold: highest counter where all lower counters are in DB (contiguous + prefix). event_handler skips counter <= threshold with an O(1) check. None + disables dedup (normal jobs, zero overhead). + persisted_counters: small set of counters above dedup_threshold that are already + in DB (the collision zone from parallel callback worker races). Bounded by + worker concurrency, not job size. None when dedup is disabled. + """ + self.instance = instance + self.job_created = str(instance.created) # stamped on every event (callback.py:160) + self.safe_env = safe_env if safe_env is not None else {} + if getattr(instance, 'spawned_by_workflow', False): + try: + self.parent_workflow_job_id = instance.get_workflow_job().id + except Exception: + pass + self.dedup_threshold = dedup_threshold + self.persisted_counters = persisted_counters + # Populate host_map from inventory hostvars using the same logic as normal job path. + # This ensures adoption respects inventory types (smart, constructed, normal), + # enabled state, and job slicing — the same way write_inventory_file() does. + if hasattr(instance, 'inventory') and instance.inventory_id: + try: + script_params = {"hostvars": True, "towervars": True} + if hasattr(instance, 'job_slice_number'): + script_params['slice_number'] = instance.job_slice_number + script_params['slice_count'] = instance.job_slice_count + script_data = instance.inventory.get_script_data(**script_params) + for hostname, hv in script_data.get('_meta', {}).get('hostvars', {}).items(): + self.host_map[hostname] = hv.get('remote_tower_id', '') + except Exception: + pass # host_map stays {}; host_id won't be set on replayed events + + @classmethod + def create_for_job(cls, instance, safe_env=None, dedup_threshold=None, persisted_counters=None): + """Factory: create and configure a RunnerCallback for a job.""" + callback = cls(model=type(instance)) + callback.configure_for_job(instance, safe_env, dedup_threshold, persisted_counters) + return callback + def event_handler(self, event_data): # # ⚠️ D-D-D-DANGER ZONE ⚠️ @@ -144,6 +197,13 @@ class RunnerCallback: # logger if event_data.get('event') == 'keepalive': return + if self.dedup_threshold is not None: + counter = event_data.get('counter') + if counter is not None: + if counter <= self.dedup_threshold: + return # contiguous safe range — all in DB, O(1) check + if self.persisted_counters and counter in self.persisted_counters: + return # collision zone — in DB but above threshold, O(small-set) check if event_data.get(self.event_data_key, None): if self.event_data_key != 'job_id': diff --git a/awx/main/tasks/host_metrics.py b/awx/main/tasks/host_metrics.py index 71572d7af9..6cfe6010bf 100644 --- a/awx/main/tasks/host_metrics.py +++ b/awx/main/tasks/host_metrics.py @@ -120,7 +120,6 @@ class HostMetricSummaryMonthlyTask: self.records_to_update = [] def execute(self): - with advisory_lock( HostMetricSummaryMonthlyTask.LOCK_KEY, lock_session_timeout_milliseconds=HostMetricSummaryMonthlyTask.LOCK_SESSION_TIMEOUT, wait=False ) as acquired: diff --git a/awx/main/tasks/jobs.py b/awx/main/tasks/jobs.py index 15fa91906f..9d8a9dfef8 100644 --- a/awx/main/tasks/jobs.py +++ b/awx/main/tasks/jobs.py @@ -46,7 +46,6 @@ from awx.main.constants import ( ) from awx.main.models import ( Instance, - Inventory, InventorySource, UnifiedJob, Job, @@ -205,6 +204,42 @@ def dispatch_waiting_jobs(binder): UnifiedJob.objects.filter(pk=uj.pk, status='waiting').update(status='running', start_args='') +def _finalize_job_run(model, pk, runner_callback, status, extra_fields=None): + """Commit terminal status, trigger notifications, and emit websocket status. + + Shared by BaseTask.run() (normal path) and adoption (reattach_to_work_unit path). + Uses duck typing on the instance to schedule dependent task/workflow managers. + """ + all_fields = runner_callback.get_delayed_update_fields() + if extra_fields: + all_fields.update(extra_fields) + instance = update_model(model, pk, status=status, select_for_update=True, **all_fields) + if not instance: + return None + if (instance.host_status_counts is not None) or (not runner_callback.wrapup_event_dispatched): + events_processed_hook(instance) + instance.websocket_emit_status(status) + + # Schedule dependent task/workflow managers via duck typing on the instance. + # This works for all job types: Job, ProjectUpdate, InventoryUpdate, etc. + instance.log_lifecycle("finalize_run") + + if hasattr(instance, 'unifiedjob_blocked_jobs') and instance.unifiedjob_blocked_jobs.exists(): + ScheduleTaskManager().schedule() + + if hasattr(instance, 'spawned_by_workflow') and instance.spawned_by_workflow: + ScheduleWorkflowManager().schedule() + + # For jobs with inventory, schedule computed fields update (RunJob-specific finalization) + if hasattr(instance, 'inventory_id') and instance.inventory_id: + try: + update_inventory_computed_fields.delay(instance.inventory_id) + except Exception: + logger.exception(f'{instance.log_format} Error scheduling inventory computed fields update') + + return instance + + class BaseTask(object): model = None event_model = None @@ -486,15 +521,11 @@ class BaseTask(object): def write_inventory_file(self, inventory, private_data_dir, file_name, script_params): script_data = inventory.get_script_data(**script_params) - for hostname, hv in script_data.get('_meta', {}).get('hostvars', {}).items(): - # maintain a list of host_name --> host_id - # so we can associate emitted events to Host objects - self.runner_callback.host_map[hostname] = hv.get('remote_tower_id', '') file_content = '#! /usr/bin/env python3\n# -*- coding: utf-8 -*-\nprint(%r)\n' % json.dumps(script_data) return self.write_private_data_file(private_data_dir, file_name, file_content, sub_dir='inventory', file_permissions=0o700) def build_inventory(self, instance, private_data_dir): - script_params = dict(hostvars=True, towervars=True) + script_params = {"hostvars": True, "towervars": True} if hasattr(instance, 'job_slice_number'): script_params['slice_number'] = instance.job_slice_number script_params['slice_count'] = instance.job_slice_count @@ -612,18 +643,6 @@ class BaseTask(object): """ instance.log_lifecycle("post_run") - def final_run_hook(self, instance, status, private_data_dir): - """ - Hook for any steps to run after job/task is marked as complete. - """ - instance.log_lifecycle("finalize_run") - - # Run task manager appropriately for speculative dependencies - if instance.unifiedjob_blocked_jobs.exists(): - ScheduleTaskManager().schedule() - if instance.spawned_by_workflow: - ScheduleWorkflowManager().schedule() - def should_use_fact_cache(self): return False @@ -663,6 +682,7 @@ class BaseTask(object): self.safe_cred_env = {} private_data_dir = None + receptor_job = None try: if self.instance.execution_environment_id is None: @@ -703,16 +723,9 @@ class BaseTask(object): self.build_extra_vars_file(self.instance, private_data_dir) args = self.build_args(self.instance, private_data_dir, passwords) env = self.build_env(self.instance, private_data_dir, private_data_files=private_data_files) - self.runner_callback.safe_env = build_safe_env(env) - - self.runner_callback.instance = self.instance - - # store a reference to the parent workflow job (if any) so we can include - # it in event data JSON - if self.instance.spawned_by_workflow: - self.runner_callback.parent_workflow_job_id = self.instance.get_workflow_job().id - - self.runner_callback.job_created = str(self.instance.created) + # Initialize common callback fields so that the normal job path exercises + # the same configuration code as adoption. + self.runner_callback.configure_for_job(self.instance, safe_env=build_safe_env(env)) credentials = self._credentials @@ -795,6 +808,13 @@ class BaseTask(object): self.unit_id = receptor_job.unit_id if not res: + # res is None when quota exceeded or other early-return condition. + # Must release work unit here before returning, or it will leak. + if receptor_job and getattr(receptor_job, 'receptor_ctl', None): + try: + receptor_job._receptor_release_work(receptor_job.receptor_ctl, 'error') + except Exception: + logger.exception(f'Failed to release work unit {getattr(receptor_job, "unit_id", "unknown")} on early return') return status = res.status @@ -832,23 +852,21 @@ class BaseTask(object): except Exception: logger.exception('{} Post run hook errored.'.format(self.instance.log_format)) - self.instance = self.update_model(pk) - self.instance = self.update_model(pk, status=status, select_for_update=True, **self.runner_callback.get_delayed_update_fields()) + self.private_data_dir = private_data_dir + try: + self.instance = _finalize_job_run(self.model, pk, self.runner_callback, status) + finally: + # Guarantee work unit release even if finalization throws + if receptor_job and getattr(receptor_job, 'receptor_ctl', None): + try: + receptor_job._receptor_release_work(receptor_job.receptor_ctl, status) + except Exception: + logger.exception(f'Failed to release work unit {getattr(receptor_job, "unit_id", "unknown")}') - # Field host_status_counts is used as a metric to check if event processing is finished - # we send notifications if it is, if not, callback receiver will send them if not self.instance: logger.error(f'Unified job pk={pk} appears to be deleted while running') return - if (self.instance.host_status_counts is not None) or (not self.runner_callback.wrapup_event_dispatched): - events_processed_hook(self.instance) - try: - self.final_run_hook(self.instance, status, private_data_dir) - except Exception: - logger.exception('{} Final run hook errored.'.format(self.instance.log_format)) - - self.instance.websocket_emit_status(status) if status != 'successful': if status == 'canceled': raise AwxTaskError.TaskCancel(self.instance, rc) @@ -1374,16 +1392,6 @@ class RunJob(SourceControlMixin, BaseTask): job_created=job.created, ) - def final_run_hook(self, job, status, private_data_dir): - super(RunJob, self).final_run_hook(job, status, private_data_dir) - try: - inventory = job.inventory - except Inventory.DoesNotExist: - pass - else: - if inventory is not None: - update_inventory_computed_fields.delay(inventory.id) - @task(queue=get_task_queuename) class RunProjectUpdate(BaseTask): @@ -1826,7 +1834,7 @@ class RunInventoryUpdate(SourceControlMixin, BaseTask): inventory_update.log_lifecycle("start_job_fact_cache") for input_inventory in inventory_update.inventory.input_inventories.all(): args.append('-i') - script_params = dict(hostvars=True, towervars=True) + script_params = {"hostvars": True, "towervars": True} source_inv_path = self.write_inventory_file(input_inventory, private_data_dir, f'hosts_{input_inventory.id}', script_params) args.append(get_incontainer_path(source_inv_path, private_data_dir)) # Include any facts from input inventories so they can be used in filters diff --git a/awx/main/tasks/receptor.py b/awx/main/tasks/receptor.py index fdab631139..168e678a8f 100644 --- a/awx/main/tasks/receptor.py +++ b/awx/main/tasks/receptor.py @@ -3,16 +3,20 @@ from base64 import b64encode from collections import namedtuple import concurrent.futures from enum import Enum +import json import logging import os import shutil import socket +import tempfile import time import yaml # Django from django.conf import settings from django.db import connections +from django.db.models import Exists, OuterRef +from django.utils.timezone import now # Runner import ansible_runner @@ -31,11 +35,13 @@ from awx.main.utils.common import ( parse_yaml_or_json, cleanup_new_process, ) -from awx.main.constants import MAX_ISOLATED_PATH_COLON_DELIMITER +from awx.main.constants import JOB_FOLDER_PREFIX, MAX_ISOLATED_PATH_COLON_DELIMITER from awx.main.tasks.signals import signal_state, signal_callback, SignalExit +from awx.main.tasks.callback import RunnerCallback from awx.main.models import Instance, InstanceLink, UnifiedJob, ReceptorAddress from awx.main.dispatch import get_task_queuename + # Receptorctl from receptorctl.socket_interface import ReceptorControl @@ -175,6 +181,37 @@ def get_receptor_ctl(config_data=None): return ReceptorControl(receptor_sockfile) +def adopt_remote_work(receptor_ctl, node, unit_id, config_data=None): + """Adopt a running work unit from a remote node, passing TLS/signwork when configured. + + PR#1564 adds adopt_work() to receptorctl. If not available (older client), falls back + to the JSON control-socket protocol directly — compatible with any PR#1564 receptor server. + + TEMPORARY: Once receptor#1564 becomes the minimum required version and requirements.txt + is updated, the fallback block (lines ~197–206) can be removed since adopt_work() will + always be available. This handles mixed-version deployments during transition. + """ + if config_data is None: + config_data = read_receptor_config() + tls_client = get_tls_client(config_data, True) + sign = work_signing_enabled(config_data) + + if hasattr(receptor_ctl, 'adopt_work'): + return receptor_ctl.adopt_work(node, unit_id, tlsclient=tls_client or None, signwork=sign) + + # TODO: Delete this fallback block once receptor#1564 is the minimum required version. + # Older receptorctl: send JSON directly so the server's InitFromJSON path handles TLS/signwork. + # The text-protocol InitFromString path does NOT support tlsclient/signwork. + command = {"command": "work", "subcommand": "adopt", "node": node, "unitid": unit_id} + if tls_client: + command["tlsclient"] = tls_client + if sign: + command["signwork"] = True + receptor_ctl.connect() + receptor_ctl.writestr(json.dumps(command) + "\n") + return receptor_ctl.read_and_parse_json() + + def find_node_in_mesh(node_name, receptor_ctl): attempts = 10 backoff = 1 @@ -406,49 +443,41 @@ class AWXReceptorJob: def run(self): # We establish a connection to the Receptor socket self.config_data = read_receptor_config() - receptor_ctl = get_receptor_ctl(self.config_data) - - res = None - try: - res = self._run_internal(receptor_ctl) - return res - finally: - status = getattr(res, 'status', 'error') - self._receptor_release_work(receptor_ctl, status) + self.receptor_ctl = get_receptor_ctl(self.config_data) + return self._run_internal(self.receptor_ctl) def _receptor_release_work(self, receptor_ctl: ReceptorControl, status: str) -> None: - """ - Releases the work unit from Receptor if certain conditions are met. - This method checks several conditions before attempting to release the work unit: - - If `self.unit_id` is `None`, the method returns immediately. - - If the `RECEPTOR_RELEASE_WORK` setting is `False`, the method returns immediately. - - If the `RECEPTOR_KEEP_WORK_ON_ERROR` setting is `True` and the status is 'error', the method returns immediately. - If none of the above conditions are met, the method attempts to release the work unit using the Receptor control command. - If an exception occurs during the release process, it logs an error message. - Args: - receptor_ctl (ReceptorControl): The Receptor control object used to issue commands. - status (str): The status of the work unit, which may affect whether it is released. - """ - if self.unit_id is None: - logger.debug("No work unit ID to release.") return if settings.RECEPTOR_RELEASE_WORK is False: - logger.debug(f"RECEPTOR_RELEASE_WORK is False, not releasing work unit {self.unit_id}.") return if settings.RECEPTOR_KEEP_WORK_ON_ERROR and status == 'error': - logger.debug(f"RECEPTOR_KEEP_WORK_ON_ERROR is True and status is 'error', not releasing work unit {self.unit_id}.") return try: - logger.debug(f"Released work unit {self.unit_id}.") receptor_ctl.simple_command(f"work release {self.unit_id}") + logger.debug(f"Released work unit {self.unit_id}.") except Exception: logger.exception(f"Error releasing work unit {self.unit_id}.") def _run_internal(self, receptor_ctl): + self._transmit_phase(receptor_ctl) + return self._process_phase(receptor_ctl) + + def _transmit_phase(self, receptor_ctl): + """Submit work to receptor and wait for the transmit thread to finish. + + Creates the receptor work unit, streams the job payload (private_data_dir + kwargs) + to the EE via a socketpair, and saves the resulting unit_id to the DB. + + After this returns, artifacts/ is deleted from private_data_dir. Fact-cache and + other input artifacts are transmitted to the EE during this phase; by the time + _process_phase() reads artifacts output, the dir is empty so there is no stale + input state. This means _process_phase() always receives a clean private_data_dir + whether called from _run_internal() or from reattach_to_work_unit(). + """ # Create a socketpair. Where the left side will be used for writing our payload # (private data dir, kwargs). The right side will be passed to Receptor for # reading. @@ -497,7 +526,20 @@ class AWXReceptorJob: if self.work_type != 'local' and os.path.exists(artifact_dir): shutil.rmtree(artifact_dir) - resultsock, resultfile = receptor_ctl.get_work_results(self.unit_id, return_socket=True, return_sockfile=True) + def _process_phase(self, receptor_ctl): + """Stream events from the receptor work unit via the ansible-runner process streamer. + + Extracted from _run_internal() so it can be called standalone for job adoption + after a same-controller restart (reattach_to_work_unit). The transmit phase + (submit_work + transmit) is not repeated — the EE is already running. + """ + resultsock = None + resultfile = None + try: + resultsock, resultfile = receptor_ctl.get_work_results(self.unit_id, return_socket=True, return_sockfile=True) + except Exception: + logger.exception(f'Failed to get work results for unit {self.unit_id}') + raise connections.close_all() @@ -516,56 +558,68 @@ class AWXReceptorJob: res = processor_future.result() except SignalExit: receptor_ctl.simple_command(f"work cancel {self.unit_id}") - resultsock.shutdown(socket.SHUT_RDWR) - resultfile.close() + if resultsock: + try: + resultsock.shutdown(socket.SHUT_RDWR) + except Exception: + pass + if resultfile: + resultfile.close() result = namedtuple('result', ['status', 'rc']) res = result('canceled', 1) finally: signal_state.raise_exception = False if res.status == 'error': - # If ansible-runner ran, but an error occured at runtime, the traceback information - # is saved via the status_handler passed in to the processor. - if 'result_traceback' in self.task.runner_callback.extra_update_fields: - return res + return self._handle_work_error(receptor_ctl, res) - try: - unit_status = receptor_ctl.simple_command(f'work status {self.unit_id}') - detail = unit_status.get('Detail', None) - state_name = unit_status.get('StateName', None) - stdout_size = unit_status.get('StdoutSize', 0) - except Exception: - detail = '' - state_name = '' - stdout_size = 0 - logger.exception(f'An error was encountered while getting status for work unit {self.unit_id}') + return res - if 'exceeded quota' in detail: - logger.warning(detail) - log_name = self.task.instance.log_format - logger.warning(f"Could not launch pod for {log_name}. Exceeded quota.") - self.task.update_model(self.task.instance.pk, status='pending') - return + def _handle_work_error(self, receptor_ctl, res): + """Handle the error path from _process_phase when the work unit status is 'error'.""" + # If ansible-runner ran, but an error occured at runtime, the traceback information + # is saved via the status_handler passed in to the processor. + if 'result_traceback' in self.task.runner_callback.extra_update_fields: + return res - try: - receptor_output = '' - if state_name == 'Failed' and self.task.runner_callback.event_ct == 0: - # if receptor work unit failed and no events were emitted, work results may - # contain useful information about why the job failed. In case stdout is - # massive, only ask for last 1000 bytes - startpos = max(stdout_size - 1000, 0) - resultsock, resultfile = receptor_ctl.get_work_results(self.unit_id, startpos=startpos, return_socket=True, return_sockfile=True) - lines = resultfile.readlines() - receptor_output = b"".join(lines).decode() - if receptor_output: - self.task.runner_callback.delay_update(result_traceback=f'Worker output:\n{receptor_output}') - elif detail: - self.task.runner_callback.delay_update(result_traceback=f'Receptor detail:\n{detail}') - else: - logger.warning(f'No result details or output from {self.task.instance.log_format}, status:\n{state_name}') - except Exception: - logger.exception(f'Work results error from job id={self.task.instance.id} work_unit={self.task.instance.work_unit_id}') - raise RuntimeError(detail) + try: + unit_status = receptor_ctl.simple_command(f'work status {self.unit_id}') + detail = unit_status.get('Detail') or '' + state_name = unit_status.get('StateName', None) + stdout_size = unit_status.get('StdoutSize', 0) + except Exception: + detail = '' + state_name = '' + stdout_size = 0 + logger.exception(f'An error was encountered while getting status for work unit {self.unit_id}') + + if 'exceeded quota' in detail: + logger.warning(detail) + log_name = self.task.instance.log_format + logger.warning(f"Could not launch pod for {log_name}. Exceeded quota.") + self.task.update_model(self.task.instance.pk, status='pending') + return None + + try: + receptor_output = '' + if state_name == 'Failed' and self.task.runner_callback.event_ct == 0: + # if receptor work unit failed and no events were emitted, work results may + # contain useful information about why the job failed. In case stdout is + # massive, only ask for last 1000 bytes + startpos = max(stdout_size - 1000, 0) + _resultsock, resultfile = receptor_ctl.get_work_results(self.unit_id, startpos=startpos, return_socket=True, return_sockfile=True) + lines = resultfile.readlines() + receptor_output = b"".join(lines).decode() + _resultsock.shutdown(socket.SHUT_RDWR) + if receptor_output: + self.task.runner_callback.delay_update(result_traceback=f'Worker output:\n{receptor_output}') + elif detail: + self.task.runner_callback.delay_update(result_traceback=f'Receptor detail:\n{detail}') + else: + logger.warning(f'No result details or output from {self.task.instance.log_format}, status:\n{state_name}') + except Exception: + logger.exception(f'Work results error from job id={self.task.instance.id} work_unit={self.task.instance.work_unit_id}') + raise RuntimeError(detail) return res @@ -799,6 +853,199 @@ RECEPTOR_CONFIG_STARTER = ( ) +class _AdoptionTask: + """Minimal task stub passed to AWXReceptorJob when called from reattach_to_work_unit. + + Stands in for BaseTask without re-executing task-manager logic. update_model is a + no-op since adoption doesn't touch scheduling fields (those are handled by the + reaper/heartbeat path that triggered the adoption). + """ + + def __init__(self, instance, runner_callback): + self.instance = instance + self.runner_callback = runner_callback + self.private_data_dir = None # Set by reattach_to_work_unit before finalization + + def build_execution_environment_params(self, _instance, _private_data_dir): + return {} + + def update_model(self, pk, **kwargs): + # Intentionally a no-op: adoption reuses _process_phase without task-manager field updates. + # Heartbeat/reaper handles scheduling state; no fields need updating here. + pass + + +def _get_adoption_exit_code(unit_status, state_name): + """Extract exit code from a finished receptor work unit status dict.""" + if 'ExitCode' in unit_status: + return unit_status['ExitCode'] + detail = unit_status.get('Detail', '') + try: + return int(detail.split()[-1]) + except (ValueError, IndexError): + return 0 if state_name == 'Succeeded' else 1 + + +def _build_adoption_callback(job, dedup_threshold, collision_zone): + """Construct a RunnerCallback for event replay during adoption.""" + return RunnerCallback.create_for_job(job, dedup_threshold=dedup_threshold, persisted_counters=collision_zone) + + +def _get_or_create_private_data_dir(job): + """Return a writable private_data_dir for the adoption process streamer. + + On same-controller restart the original tempdir may be gone (e.g. on OCP where /tmp + is wiped on pod restart). ansible-runner's process streamer only needs a writable root + directory — it creates artifacts/job_events/ itself. + + Uses AWX_ISOLATION_BASE_PATH (the same root as all other job working dirs) so that + sysadmin-configured isolation paths (e.g. persistent volumes on OCP) are respected. + """ + return tempfile.mkdtemp( + prefix=JOB_FOLDER_PREFIX % job.pk + 'adoption_', + dir=settings.AWX_ISOLATION_BASE_PATH, + ) + + +def _finalize_adopted_job(job, callback, exit_code, process_phase_failed): + """Commit terminal status for an adopted job via the shared _finalize_job_run path. + + The shared finalization function uses duck typing to schedule task/workflow managers + for speculative dependencies and workflow jobs. + + Guards before calling: if awx_receptor_workunit_reaper already committed the final + status, there is nothing left to do. + """ + from awx.main.tasks.jobs import _finalize_job_run + + job.refresh_from_db(fields=['status']) + if job.status != 'running': + return + + final_status = 'successful' if exit_code == 0 else 'failed' + finished_at = now() + extra = {'finished': finished_at} + if job.started: + extra['elapsed'] = (finished_at - job.started).total_seconds() + + _finalize_job_run(type(job), job.pk, callback, final_status, extra_fields=extra) + + label = 'exit_code (process phase raised)' if process_phase_failed else 'adoption' + logger.info(f'Job {job.id} finalized via {label}: {final_status}') + + +def _compute_adoption_dedup(job): + """Return (safe_threshold, collision_zone) for counter-skip dedup during adoption. + + Hybrid approach — memory is O(1) + O(worker_count), never O(total events): + + safe_threshold: highest counter where all lower counters are also in DB (contiguous + prefix). Events <= this are skipped with a single integer comparison. + + collision_zone: small set of counters above safe_threshold that ARE in DB. These + exist because parallel callback workers can commit a higher-counter event before + a lower-counter one. Bounded by JOB_EVENT_WORKERS × batch size, typically < 20 + regardless of total job event count. + """ + next_ctr = job.get_event_queryset().filter(counter=OuterRef('counter') + 1) + gap_event = job.get_event_queryset().annotate(has_next=Exists(next_ctr)).filter(has_next=False).order_by('counter').first() + safe_threshold = gap_event.counter if gap_event else 0 + + cap = settings.JOB_EVENT_WORKERS * settings.JOB_EVENT_CALLBACK_BUFFER_SIZE + collision_zone_list = list(job.get_event_queryset().filter(counter__gt=safe_threshold).values_list('counter', flat=True)) + + if len(collision_zone_list) > cap: + logger.warning( + f'Job {job.id}: collision_zone has {len(collision_zone_list)} events above safe_threshold, ' + f'exceeds dedup cap of {cap}. Events beyond cap may be re-processed if replayed. ' + f'Consider increasing JOB_EVENT_CALLBACK_BUFFER_SIZE or reducing parallel callback workers.' + ) + + collision_zone = set(collision_zone_list[:cap]) + return safe_threshold, collision_zone + + +def reattach_to_work_unit(job, receptor_ctl): + """Reconnect to a receptor work unit and stream events in real-time until it completes. + + Reconstructs the minimal process-phase context from the DB job record, then calls + _process_phase which blocks until the work unit finishes — streaming events live as + the EE generates them. Dedup (safe_threshold + collision_zone) skips events already + in DB so replay from startpos=0 is safe. + + Intended to be called from adopt_job_async (a background task) so the caller is not + blocked. Cross-controller path (node=execution_node) is deferred to AAP-89602. + """ + unit_id = job.work_unit_id + + # Check state for logging — no longer a gate. We stream regardless. + try: + unit_status = receptor_ctl.simple_command(f'work status {unit_id}') + except Exception: + logger.warning(f'Cannot get receptor status for work unit {unit_id} (job {job.id}), deferring adoption') + return False + + state_name = unit_status.get('StateName', '') + logger.info(f'Adopting job {job.id}: unit {unit_id} in state {state_name!r}, starting real-time streaming') + + # Pending/Running — EE not yet finished; defer to next heartbeat. + if state_name in ('Pending', 'Running'): + logger.info(f'Job {job.id}: unit {unit_id} in state {state_name!r}, deferring to next heartbeat') + return False + + safe_threshold, collision_zone = _compute_adoption_dedup(job) + max_counter = max(collision_zone) if collision_zone else safe_threshold + logger.info( + f'Job {job.id}: safe_threshold={safe_threshold} collision_zone_size={len(collision_zone)} ' + f'(max counter={max_counter}), replaying from startpos=0 with counter-skip' + ) + + callback = _build_adoption_callback(job, safe_threshold, collision_zone) + + private_data_dir = _get_or_create_private_data_dir(job) + adoption_task = _AdoptionTask(job, callback) + adoption_task.private_data_dir = private_data_dir # Available to final_run_hook + receptor_job = AWXReceptorJob(adoption_task, {'private_data_dir': private_data_dir}) + receptor_job.unit_id = unit_id + + process_phase_failed = False + res = None + try: + res = receptor_job._process_phase(receptor_ctl) # blocks until unit completes + except Exception: + logger.exception(f'Adoption process phase failed for job {job.id} (unit={unit_id})') + process_phase_failed = True + + # Finalize status to DB first, then release the work unit — matching the ordering + # in BaseTask.run() where release follows _finalize_job_run. + # Keep private_data_dir available through finalization for final_run_hook. + # + # _process_phase -> _handle_work_error may return None for 'exceeded quota' where + # the job is already set to 'pending' — don't finalize, let the next dispatch handle it. + try: + if res is not None: + exit_code = 0 if getattr(res, 'status', '') == 'successful' else 1 + _finalize_adopted_job(job, callback, exit_code, process_phase_failed) + elif not process_phase_failed: + logger.info(f'Job {job.id}: adoption deferred (handled in _handle_work_error)') + else: + # _process_phase raised — fall back to work status for exit code + try: + unit_status = receptor_ctl.simple_command(f'work status {unit_id}') + state_name = unit_status.get('StateName', '') + except Exception: + logger.debug(f'Could not get final status for {unit_id} after process phase failure') + unit_status = {} + state_name = '' + exit_code = _get_adoption_exit_code(unit_status, state_name) + _finalize_adopted_job(job, callback, exit_code, process_phase_failed) + finally: + receptor_job._receptor_release_work(receptor_ctl, getattr(res, 'status', 'error')) + shutil.rmtree(private_data_dir, ignore_errors=True) + + return True + + def should_update_config(new_config): ''' checks that the list of instances matches the list of diff --git a/awx/main/tasks/system.py b/awx/main/tasks/system.py index fc2a04fa93..47b3823a1e 100644 --- a/awx/main/tasks/system.py +++ b/awx/main/tasks/system.py @@ -36,7 +36,9 @@ from django.conf import settings from django.contrib.auth.models import User from django.core.cache import cache from django.core.exceptions import ObjectDoesNotExist +from django.contrib.contenttypes.models import ContentType from django.db import DatabaseError, IntegrityError, connection, transaction +from django.db.models import Max, Q from django.db.models.fields.related import ForeignKey from django.db.models.query import QuerySet from django.utils.encoding import smart_str @@ -65,6 +67,7 @@ from awx.main.models import ( SmartInventoryMembership, TowerScheduleState, UnifiedJob, + WorkflowJob, convert_jsonfields, ) from awx.main.models.credential import CredentialType @@ -73,6 +76,7 @@ from awx.main.tasks.host_indirect import save_indirect_host_entries from awx.main.tasks.receptor import ( administrative_workunit_reaper, get_receptor_ctl, + reattach_to_work_unit, worker_cleanup, worker_info, write_receptor_config, @@ -149,7 +153,10 @@ def _run_dispatch_startup_common(): # apply_cluster_membership_policies() cluster_node_heartbeat(None) - reaper.startup_reaping() + # Safety net: reap undispatched startup jobs even if cluster_node_heartbeat returned early + # (receptor unavailable, rejoining cluster). _process_startup_jobs handles adoption; this + # handles the reap-only case. reap_job() is idempotent — already-reaped jobs are skipped. + _startup_reap_undispatched(settings.CLUSTER_HOST_ID) m = DispatcherMetrics() m.reset_values() @@ -661,8 +668,9 @@ def cluster_node_heartbeat(binder): Uses Control API to get running tasks. """ - # Run common instance management logic - this_inst, instance_list, lost_instances = _heartbeat_instance_management() + # Run common instance management logic — ctl is the same receptor connection used for + # mesh status; we reuse it for the job processing loop to avoid a second socket open. + this_inst, instance_list, lost_instances, _ctl = _heartbeat_instance_management() if this_inst is None: return # Early return case from instance management @@ -672,19 +680,24 @@ def cluster_node_heartbeat(binder): # Handle lost instances _heartbeat_handle_lost_instances(lost_instances, this_inst) - # Get running tasks using dispatcherd API if binder is None: + # Startup: one loop over all running jobs on this instance. + # Dispatched jobs are handed to adopt_job_async; undispatched jobs are reaped. + _process_startup_jobs(this_inst) logger.debug("Heartbeat finished in startup.") return + + # Periodic heartbeat: get running tasks from dispatcherd, then process orphaned jobs. active_task_ids = _get_active_task_ids_from_dispatcherd(binder) if active_task_ids is None: logger.warning("No active task IDs retrieved from dispatcherd, skipping reaper") return # Failed to get task IDs, don't attempt reaping - # Run local reaper using tasks from dispatcherd - ref_time = now() # No dispatch_time in dispatcherd version - logger.debug(f"Running reaper with {len(active_task_ids)} excluded UUIDs") - reaper.reap(instance=this_inst, excluded_uuids=active_task_ids, ref_time=ref_time) + # One loop over all orphaned running jobs — adopt dispatched, reap undispatched. + ref_time = now() + logger.debug(f"Running job processing loop with {len(active_task_ids)} excluded UUIDs") + _process_running_jobs(this_inst, active_task_ids, ref_time) + # If waiting jobs are hanging out, resubmit them if UnifiedJob.objects.filter(controller_node=settings.CLUSTER_HOST_ID, status='waiting').exists(): from awx.main.tasks.jobs import dispatch_waiting_jobs @@ -772,7 +785,7 @@ def _heartbeat_instance_management(): if this_inst: this_inst.local_health_check() this_inst.mark_offline(errors='Receptor not available') - return None, None, None + return None, None, None, None try: mesh_status = ctl.simple_command('status') @@ -795,7 +808,7 @@ def _heartbeat_instance_management(): this_inst.local_health_check() if startup_event and this_inst.capacity != 0: logger.warning(f'Rejoining the cluster as instance {this_inst.hostname}. Prior last_seen {last_last_seen}') - return None, None, None # Early return case + return None, None, None, None # Early return case elif not last_last_seen: logger.warning(f'Instance does not have recorded last_seen, updating to {nowtime}') elif (nowtime - last_last_seen) > timedelta(seconds=settings.CLUSTER_NODE_HEARTBEAT_PERIOD + 2): @@ -808,15 +821,15 @@ def _heartbeat_instance_management(): this_inst.local_health_check() else: logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID)) - return None, None, None + return None, None, None, None if lost_instances and not _mesh_all_ready_nodes_visible(mesh_status): # Mesh gate blocks cleanup, but execution and hop nodes can still be reaped # (they don't depend on mesh consensus). Defer only control nodes. execution_hop_lost = [inst for inst in lost_instances if inst.node_type in ('execution', 'hop')] - return this_inst, instance_list, execution_hop_lost + return this_inst, instance_list, execution_hop_lost, ctl - return this_inst, instance_list, lost_instances + return this_inst, instance_list, lost_instances, ctl def _heartbeat_check_versions(this_inst, instance_list): @@ -841,9 +854,17 @@ def _heartbeat_check_versions(this_inst, instance_list): def _reap_and_mark_lost_instance(other_inst): """Reap a lost instance's running jobs and mark it offline (or deprovision it).""" try: - # Only reap jobs never dispatched to receptor; dispatched jobs are left for adoption (AAP-89602) - explanation = "Job reaped due to instance shutdown" - reaper.reap(other_inst, job_explanation=explanation) + workflow_ctype_id = ContentType.objects.get_for_model(WorkflowJob).id + running_jobs = list( + UnifiedJob.objects.filter( + Q(execution_node=other_inst.hostname) | Q(controller_node=other_inst.hostname), + status='running', + ).exclude(polymorphic_ctype_id=workflow_ctype_id) + ) + for j in running_jobs: + # AAP-89602: cross-controller adoption for dispatched jobs will be added here + # once ansible/receptor#1564 merges. Both paths reap identically for now. + reaper.reap_job(j, 'failed', job_explanation='Job reaped due to instance shutdown') # Any jobs that were waiting to be processed by this node will be handed back to task manager UnifiedJob.objects.filter(status='waiting', controller_node=other_inst.hostname).update(status='pending', controller_node='', execution_node='') except Exception: @@ -884,6 +905,135 @@ def _heartbeat_handle_lost_instances(lost_instances, this_inst): _reap_and_mark_lost_instance(other_inst) +def _startup_reap_undispatched(hostname): + """Reap undispatched running jobs at startup using only the hostname — no receptor needed. + + Called unconditionally from _run_dispatch_startup_common so undispatched jobs are always + reaped, even when cluster_node_heartbeat() returns early (receptor unavailable, rejoining + cluster). reap_job() is idempotent: jobs already handled by _process_startup_jobs are + skipped via the running-status check. + """ + workflow_ctype_id = ContentType.objects.get_for_model(WorkflowJob).id + jobs = list( + UnifiedJob.objects.filter( + status='running', + controller_node=hostname, + ) + .filter(Q(work_unit_id='') | Q(work_unit_id=None)) + .exclude(polymorphic_ctype_id=workflow_ctype_id) + ) + job_ids = [j.id for j in jobs] + for j in jobs: + reaper.reap_job( + j, + 'failed', + job_explanation='Task was marked as running at system start up. The system must have not shut down properly, so it has been marked as failed.', + ) + if job_ids: + logger.error(f'Unified jobs {job_ids} were reaped on dispatch startup') + + +def _process_startup_jobs(this_inst): + """On controller restart, process all running jobs owned by this instance in one loop. + + Dispatched jobs (work_unit_id set) are handed off to adopt_job_async, which streams + events in real-time in a background task. Undispatched jobs are reaped immediately. + + Same-controller only. Cross-controller adoption (jobs from a dead peer controller) is + deferred to AAP-89602 and requires ansible/receptor#1564. + """ + workflow_ctype_id = ContentType.objects.get_for_model(WorkflowJob).id + jobs = list(UnifiedJob.objects.filter(status='running', controller_node=this_inst.hostname).exclude(polymorphic_ctype_id=workflow_ctype_id)) + if not jobs: + return + + reaped_ids = [] + for j in jobs: + try: + if j.work_unit_id: + obj, _ = adopt_job_async.apply_async(args=[j.id], queue=get_task_queuename()) + UnifiedJob.objects.filter(pk=j.id).update(celery_task_id=obj['uuid']) + else: + reaped_ids.append(j.id) + reaper.reap_job( + j, + 'failed', + job_explanation='Task was marked as running at system start up. The system must have not shut down properly, so it has been marked as failed.', + ) + except Exception: + logger.exception(f'Failed processing job {j.id} in startup job loop') + if reaped_ids: + logger.error(f'Unified jobs {reaped_ids} were reaped on dispatch startup') + + +def _process_running_jobs(this_inst, active_task_ids, ref_time): + """On each heartbeat, process running jobs that dispatcherd is no longer tracking. + + Jobs still in active_task_ids are legitimately running — leave them alone. + For orphaned jobs (not in active_task_ids): + - Dispatched (work_unit_id set, controller_node=this_inst) → adopt_job_async (real-time streaming). + - Undispatched or not owned by this controller → reap. + + Cross-controller adoption (jobs from a dead peer controller) is deferred to AAP-89602. + """ + workflow_ctype_id = ContentType.objects.get_for_model(WorkflowJob).id + base_q = Q(status='running') & (Q(execution_node=this_inst.hostname) | Q(controller_node=this_inst.hostname)) + base_q &= ~Q(polymorphic_ctype_id=workflow_ctype_id) + if active_task_ids: + base_q &= ~Q(celery_task_id__in=active_task_ids) + if ref_time: + base_q &= Q(started__lte=ref_time) + jobs = list(UnifiedJob.objects.filter(base_q)) + if not jobs: + return + + for j in jobs: + try: + if j.work_unit_id and j.controller_node == this_inst.hostname: + obj, _ = adopt_job_async.apply_async(args=[j.id], queue=get_task_queuename()) + # Record the adoption task UUID so subsequent heartbeats see this job + # as active and skip it — preventing redundant re-adoption dispatches. + UnifiedJob.objects.filter(pk=j.id).update(celery_task_id=obj['uuid']) + else: + reaper.reap_job(j, 'failed') + except Exception: + logger.exception(f'Failed processing job {j.id} in heartbeat job loop') + + +@task(queue=get_task_queuename, timeout=3600 * 2, on_duplicate='discard') +def adopt_job_async(job_id): + """Adopt a single orphaned job in a background task, streaming events in real-time. + + Called from _process_startup_jobs and _process_running_jobs via apply_async so the + heartbeat returns immediately. on_duplicate='discard' ensures only one adoption runs + per job across heartbeat cycles. + """ + job = UnifiedJob.objects.filter(id=job_id, status='running').first() + if not job: + logger.debug(f'adopt_job_async: job {job_id} is no longer running, skipping') + return + + adoption_timeout = settings.HADR_JOB_ADOPTION_TIMEOUT + timeout_cutoff = now() - timedelta(seconds=adoption_timeout) + last_event_time = job.get_event_queryset().aggregate(Max('created'))['created__max'] + orphaned_since = last_event_time or job.started + if orphaned_since and orphaned_since < timeout_cutoff: + logger.error(f'Job {job.id} (unit={job.work_unit_id}) orphaned for >{adoption_timeout}s, failing') + reaper.reap_job(job, 'failed', job_explanation='Job exceeded HADR_JOB_ADOPTION_TIMEOUT during controller restart') + return + + receptor_ctl = get_receptor_ctl() + try: + reattach_to_work_unit(job, receptor_ctl) + except Exception: + logger.exception(f'adopt_job_async: adoption failed for job {job.id} (unit={job.work_unit_id})') + finally: + try: + receptor_ctl.close() + except Exception: + pass + + @task(queue=get_task_queuename, timeout=1800, on_duplicate='queue_one') def awx_receptor_workunit_reaper(): """ diff --git a/awx/main/tests/functional/tasks/test_tasks_jobs.py b/awx/main/tests/functional/tasks/test_tasks_jobs.py index ffd5fce4e8..6aeb32dcc8 100644 --- a/awx/main/tests/functional/tasks/test_tasks_jobs.py +++ b/awx/main/tests/functional/tasks/test_tasks_jobs.py @@ -56,3 +56,89 @@ def test_runjob_run_can_accept_waiting_status(jt_linked, mocker): mock_pre_run.assert_called_once() assert status_at_pre_run == 'running' + + +@pytest.mark.django_db +def test_finalize_job_run_with_real_job_instance(jt_linked, mocker): + """Test that _finalize_job_run executes the real finalization path with actual Job instance.""" + from awx.main.tasks.jobs import _finalize_job_run + from awx.main.tasks.callback import RunnerCallback + + job = jt_linked.create_unified_job() + job.status = 'running' + job.save() + + # Create a real callback + callback = RunnerCallback(model=Job) + callback.event_ct = 5 + callback.host_status_counts = {'ok': 1, 'failed': 0} # Non-None triggers events_processed_hook + + # Mock the hooks that would be called + mocker.patch('awx.main.tasks.jobs.events_processed_hook') + mocker.patch('awx.main.tasks.jobs.ScheduleTaskManager') + mocker.patch('awx.main.tasks.jobs.ScheduleWorkflowManager') + + # Call the real finalization function + result = _finalize_job_run(Job, job.id, callback, 'successful') + + # Verify the job was updated with successful status + job.refresh_from_db() + assert job.status == 'successful' + assert result.id == job.id + + +@pytest.mark.django_db +def test_finalize_job_run_with_blocked_jobs(jt_linked, mocker): + """Test that _finalize_job_run calls ScheduleTaskManager for jobs with blocked dependencies.""" + from awx.main.tasks.jobs import _finalize_job_run + from awx.main.tasks.callback import RunnerCallback + + job = jt_linked.create_unified_job() + job.status = 'running' + job.save() + + # Create another job that depends on this one + blocked_job = jt_linked.create_unified_job() + blocked_job.status = 'blocked' + blocked_job.save() + job.unifiedjob_blocked_jobs.add(blocked_job) + + callback = RunnerCallback(model=Job) + callback.event_ct = 0 + + # Mock the hooks + mock_schedule_task = mocker.patch('awx.main.tasks.jobs.ScheduleTaskManager') + mocker.patch('awx.main.tasks.jobs.events_processed_hook') + mocker.patch('awx.main.tasks.jobs.ScheduleWorkflowManager') + + # Call finalization + _finalize_job_run(Job, job.id, callback, 'successful') + + # Verify ScheduleTaskManager was called + mock_schedule_task.return_value.schedule.assert_called_once() + + +@pytest.mark.django_db +def test_finalize_job_run_calls_websocket_emit(jt_linked, mocker): + """Test that websocket status is emitted after finalization.""" + from awx.main.tasks.jobs import _finalize_job_run + from awx.main.tasks.callback import RunnerCallback + + job = jt_linked.create_unified_job() + job.status = 'running' + job.save() + + callback = RunnerCallback(model=Job) + callback.event_ct = 0 + + # Mock websocket_emit_status + mocker.patch('awx.main.tasks.jobs.events_processed_hook') + mocker.patch('awx.main.tasks.jobs.ScheduleTaskManager') + mocker.patch('awx.main.tasks.jobs.ScheduleWorkflowManager') + mock_websocket = mocker.patch.object(type(job), 'websocket_emit_status') + + # Call finalization with failed status + _finalize_job_run(Job, job.id, callback, 'failed') + + # Verify websocket was called with correct status + mock_websocket.assert_called_with('failed') diff --git a/awx/main/tests/functional/tasks/test_tasks_system.py b/awx/main/tests/functional/tasks/test_tasks_system.py index 66da63cac7..519310f78a 100644 --- a/awx/main/tests/functional/tasks/test_tasks_system.py +++ b/awx/main/tests/functional/tasks/test_tasks_system.py @@ -6,6 +6,7 @@ import tempfile import shutil from contextlib import contextmanager from unittest import mock +from unittest.mock import MagicMock, patch import pytest @@ -20,12 +21,16 @@ from awx.main.tasks.system import ( _reap_and_mark_lost_instance, inspect_execution_and_hop_nodes, _heartbeat_instance_management, + _process_startup_jobs, + _process_running_jobs, + _startup_reap_undispatched, ) +from awx.main.dispatch.reaper import reap from awx.main.management.commands.dispatcherd import Command from django.db import DatabaseError from django.utils.timezone import now, timedelta -from awx.main.models import Instance, Inventory, Job, Organization, ReceptorAddress, InstanceLink +from awx.main.models import Instance, Inventory, Job, Organization, ReceptorAddress, InstanceLink, WorkflowJob from awx.main.models.inventory import Group, Host @@ -395,8 +400,9 @@ class TestReapAndMarkLostInstance: def test_reap_exception_does_not_prevent_mark_offline(self, settings): settings.AWX_AUTO_DEPROVISION_INSTANCES = False inst = self._inst() + Job.objects.create(controller_node=inst.hostname, status='running', work_unit_id=None) with mock.patch('awx.main.tasks.system.reaper') as mock_reaper: - mock_reaper.reap.side_effect = Exception('receptor timeout') + mock_reaper.reap_job.side_effect = Exception('receptor timeout') _reap_and_mark_lost_instance(inst) inst.refresh_from_db() assert inst.node_state == Instance.States.UNAVAILABLE @@ -543,7 +549,7 @@ def test_heartbeat_defers_lost_instances_when_mesh_gate_blocks(settings): mock.patch('awx.main.tasks.system.inspect_execution_and_hop_nodes'), mock.patch.object(Instance, 'local_health_check'), ): - _, _, lost_result = _heartbeat_instance_management() + _, _, lost_result, _ = _heartbeat_instance_management() assert lost_result == [] @@ -595,7 +601,7 @@ def test_kubernetes_passes_mesh_gate_with_empty_routing(settings): mock.patch('awx.main.tasks.system.inspect_execution_and_hop_nodes'), mock.patch.object(Instance, 'local_health_check'), ): - _, _, lost_result = _heartbeat_instance_management() + _, _, lost_result, _ = _heartbeat_instance_management() # K8s bypasses the gate — lost instances are reaped even with empty routing assert len(lost_result) == 1 @@ -637,7 +643,7 @@ def test_mesh_gate_defers_control_but_cleans_execution_hop(settings): mock.patch('awx.main.tasks.system.inspect_execution_and_hop_nodes'), mock.patch.object(Instance, 'local_health_check'), ): - _, _, lost_result = _heartbeat_instance_management() + _, _, lost_result, _ = _heartbeat_instance_management() # Execution and hop nodes should be in lost_result (reaped immediately) # Control node should be deferred (not in lost_result) @@ -649,7 +655,7 @@ def test_mesh_gate_defers_control_but_cleans_execution_hop(settings): @pytest.mark.django_db def test_heartbeat_marks_offline_when_receptor_unavailable(settings): - """FileNotFoundError from get_receptor_ctl → this_inst marked offline, returns (None, None, None).""" + """FileNotFoundError from get_receptor_ctl → this_inst marked offline, returns (None, None, None, None).""" settings.CLUSTER_HOST_ID = 'ctrl-0' settings.AWX_AUTO_DEPROVISION_INSTANCES = False @@ -663,6 +669,719 @@ def test_heartbeat_marks_offline_when_receptor_unavailable(settings): ): result = _heartbeat_instance_management() - assert result == (None, None, None) + assert result == (None, None, None, None) this_inst.refresh_from_db() assert this_inst.node_state == Instance.States.UNAVAILABLE + + +# ── AAP-89607: unified startup job processing ───────────────────────────────── + + +@pytest.mark.django_db +def test_process_startup_jobs_skips_dispatched_job(me_inst, settings): + """_process_startup_jobs() must not reap jobs that have a work_unit_id (dispatched to receptor). + + Dispatched jobs are handed to adopt_job_async for background streaming adoption. + """ + dispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='abc12345') + with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt: + mock_adopt.apply_async = MagicMock() + _process_startup_jobs(me_inst) + dispatched.refresh_from_db() + assert dispatched.status == 'running', 'dispatched job was wrongly reaped by _process_startup_jobs()' + mock_adopt.apply_async.assert_called_once() + + +@pytest.mark.django_db +def test_process_startup_jobs_reaps_undispatched_job(me_inst, settings): + """_process_startup_jobs() reaps jobs with no work_unit_id (never dispatched to receptor).""" + undispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=None) + _process_startup_jobs(me_inst) + undispatched.refresh_from_db() + assert undispatched.status == 'failed', 'undispatched job should have been reaped' + + +@pytest.mark.django_db +def test_reap_reaps_dispatched_jobs(me_inst): + """reap() reaps dispatched jobs (no undispatched_only filter).""" + dispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-xyz') + reap(instance=me_inst) + dispatched.refresh_from_db() + assert dispatched.status == 'failed', 'reap() should reap dispatched jobs' + + +# ── AAP-89607: adoption loop ────────────────────────────────────────────────── + + +@pytest.mark.django_db +def test_startup_no_op_when_no_jobs(me_inst): + """_process_startup_jobs() is a no-op when there are no running jobs.""" + with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt: + mock_adopt.apply_async = MagicMock() + _process_startup_jobs(me_inst) + mock_adopt.apply_async.assert_not_called() + + +@pytest.mark.django_db +def test_adoption_skips_still_running_work_unit(me_inst): + """Startup dispatches adopt_job_async for a still-running job — no inline blocking.""" + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='running-unit') + + with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt: + mock_adopt.apply_async = MagicMock() + _process_startup_jobs(me_inst) + + mock_adopt.apply_async.assert_called_once_with(args=[job.id], queue=mock_adopt.apply_async.call_args[1]['queue']) + job.refresh_from_db() + assert job.status == 'running', 'job must not be reaped — adoption deferred to background task' + + +@pytest.mark.django_db +def test_adoption_timeout_fails_job(me_inst, settings): + """Jobs orphaned longer than HADR_JOB_ADOPTION_TIMEOUT are reaped by adopt_job_async. + + Timeout is measured from the last event received, not from job.started. + """ + settings.HADR_JOB_ADOPTION_TIMEOUT = 3600 + from awx.main.tasks.system import adopt_job_async + + job = Job.objects.create( + controller_node=me_inst.hostname, + status='running', + work_unit_id='old-unit', + started=now() - timedelta(seconds=7200), + ) + with patch('awx.main.tasks.system.reattach_to_work_unit') as mock_reattach: + adopt_job_async(job.id) + mock_reattach.assert_not_called() + job.refresh_from_db() + assert job.status == 'failed', 'timed-out job should be reaped by adopt_job_async' + assert 'HADR_JOB_ADOPTION_TIMEOUT' in job.job_explanation + + +@pytest.mark.django_db +def test_adoption_timeout_spares_long_running_job_with_recent_events(me_inst, settings): + """Long-running jobs are NOT killed if events arrived recently. + + A job started 2 hours ago but with events 5 minutes ago (brief outage) must not be failed. + Timeout is measured from last_event.created, not job.started. + """ + from awx.main.models import JobEvent + from awx.main.tasks.system import adopt_job_async + + settings.HADR_JOB_ADOPTION_TIMEOUT = 3600 + job = Job.objects.create( + controller_node=me_inst.hostname, + status='running', + work_unit_id='long-running-unit', + started=now() - timedelta(seconds=7200), + ) + JobEvent.objects.create(job=job, counter=10, event='runner_on_ok', job_created=job.created) + JobEvent.objects.filter(job=job).update(created=now() - timedelta(seconds=300)) + + with patch('awx.main.tasks.system.get_receptor_ctl'), patch('awx.main.tasks.system.reattach_to_work_unit') as mock_reattach: + adopt_job_async(job.id) + mock_reattach.assert_called_once() + job.refresh_from_db() + assert job.status == 'running', 'Long-running job with recent events must not be failed' + + +@pytest.mark.django_db +def test_adoption_finalizes_successful_job(me_inst): + """reattach_to_work_unit finalizes a job as successful when ExitCode=0.""" + from awx.main.tasks.receptor import reattach_to_work_unit + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='done-unit') + ctl = MagicMock() + ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0} + + with ( + patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls, + patch('awx.main.tasks.callback.RunnerCallback'), + ): + mock_instance = MagicMock() + mock_instance._process_phase.return_value = MagicMock(status='successful', rc=0) + mock_job_cls.return_value = mock_instance + + reattach_to_work_unit(job, ctl) + + job.refresh_from_db() + assert job.status == 'successful' + + +@pytest.mark.django_db +def test_adoption_finalizes_failed_job(me_inst): + """reattach_to_work_unit finalizes a job as failed when ExitCode=1.""" + from awx.main.tasks.receptor import reattach_to_work_unit + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='failed-unit') + ctl = MagicMock() + ctl.simple_command.return_value = {'StateName': 'Failed', 'ExitCode': 1} + + with ( + patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls, + patch('awx.main.tasks.callback.RunnerCallback'), + ): + mock_instance = MagicMock() + mock_instance._process_phase.return_value = MagicMock(status='failed', rc=1) + mock_job_cls.return_value = mock_instance + + reattach_to_work_unit(job, ctl) + + job.refresh_from_db() + assert job.status == 'failed' + + +@pytest.mark.django_db +def test_process_running_jobs_adopts_dispatched_skips_active(me_inst): + """_process_running_jobs() dispatches adopt_job_async for orphaned jobs, skips active ones.""" + active_job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='active-unit', celery_task_id='active-uuid') + orphaned_dispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='orphaned-unit', celery_task_id='orphan-uuid') + + dispatched_ids = [] + with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt: + mock_adopt.apply_async = MagicMock(side_effect=lambda args, **kw: dispatched_ids.append(args[0])) + _process_running_jobs(me_inst, active_task_ids={'active-uuid'}, ref_time=None) + + assert orphaned_dispatched.id in dispatched_ids, 'orphaned dispatched job should be queued for adoption' + assert active_job.id not in dispatched_ids, 'active job should not be touched' + + +@pytest.mark.django_db +def test_process_running_jobs_reaps_undispatched(me_inst): + """_process_running_jobs() reaps undispatched orphaned jobs.""" + undispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=None, celery_task_id='orphan-undispatched') + _process_running_jobs(me_inst, active_task_ids=set(), ref_time=None) + undispatched.refresh_from_db() + assert undispatched.status == 'failed', 'undispatched orphaned job should be reaped' + + +@pytest.mark.django_db +def test_process_running_jobs_noop_when_no_orphaned_jobs(me_inst): + """_process_running_jobs() is a no-op (early return) when all running jobs are active.""" + active_job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='u1', celery_task_id='active-uuid') + with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt: + mock_adopt.apply_async = MagicMock() + _process_running_jobs(me_inst, active_task_ids={'active-uuid'}, ref_time=None) + mock_adopt.apply_async.assert_not_called() + active_job.refresh_from_db() + assert active_job.status == 'running' + + +@pytest.mark.django_db +def test_process_running_jobs_ref_time_filter(me_inst): + """_process_running_jobs() excludes jobs started after ref_time.""" + future_job = Job.objects.create( + controller_node=me_inst.hostname, + status='running', + work_unit_id=None, + celery_task_id='future-uuid', + started=now() + timedelta(seconds=60), + ) + ref_time = now() + with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt: + mock_adopt.apply_async = MagicMock() + _process_running_jobs(me_inst, active_task_ids=set(), ref_time=ref_time) + mock_adopt.apply_async.assert_not_called() + future_job.refresh_from_db() + assert future_job.status == 'running', 'job started after ref_time should not be touched' + + +@pytest.mark.django_db +def test_process_running_jobs_exception_does_not_abort_loop(me_inst): + """Exception in _process_running_jobs job loop is caught — remaining jobs still processed.""" + job1 = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-a', celery_task_id='uuid-a') + job2 = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-b', celery_task_id='uuid-b') + + dispatched = [] + + def side_effect(args, **kw): + if args[0] == job1.id: + raise RuntimeError('simulated failure') + dispatched.append(args[0]) + + with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt: + mock_adopt.apply_async = MagicMock(side_effect=side_effect) + _process_running_jobs(me_inst, active_task_ids=set(), ref_time=None) + + assert job2.id in dispatched, 'second job must be processed even though first raised' + + +@pytest.mark.django_db +def test_process_startup_jobs_exception_does_not_abort_loop(me_inst, settings): + """Exception in _process_startup_jobs job loop is caught — remaining jobs still processed.""" + settings.HADR_JOB_ADOPTION_TIMEOUT = 3600 + job1 = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-err') + job2 = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=None) + + def side_effect(args, **kw): + if args[0] == job1.id: + raise RuntimeError('simulated failure') + + with patch('awx.main.tasks.system.adopt_job_async') as mock_adopt: + mock_adopt.apply_async = MagicMock(side_effect=side_effect) + _process_startup_jobs(me_inst) + + job2.refresh_from_db() + assert job2.status == 'failed', 'undispatched job2 must still be reaped after job1 exception' + + +@pytest.mark.django_db +def test_startup_reap_undispatched_reaps_undispatched_leaves_dispatched(me_inst): + """_startup_reap_undispatched() reaps jobs with no work_unit_id, leaves dispatched jobs alone. + + This is the unconditional safety net called from _run_dispatch_startup_common for cases + where cluster_node_heartbeat returns early (receptor unavailable, rejoining cluster). + """ + undispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id=None) + dispatched = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='abc-unit') + _startup_reap_undispatched(me_inst.hostname) + undispatched.refresh_from_db() + dispatched.refresh_from_db() + assert undispatched.status == 'failed', 'undispatched job should be reaped by safety net' + assert dispatched.status == 'running', 'dispatched job must not be reaped (adoption loop handles it)' + + +@pytest.mark.django_db +def test_startup_reap_undispatched_no_op_when_no_jobs(me_inst): + """_startup_reap_undispatched() is a no-op when no undispatched running jobs exist.""" + _startup_reap_undispatched(me_inst.hostname) # should not raise + + +@pytest.mark.django_db +def test_startup_reap_undispatched_skips_workflow_jobs(me_inst): + """_startup_reap_undispatched() must not reap WorkflowJobs. + + WorkflowJob has work_unit_id=None (it never owns a receptor work unit) but must not + be reaped on controller restart — it coordinates via its node jobs, not receptor directly. + """ + wfj = WorkflowJob.objects.create(status='running', controller_node=me_inst.hostname) + _startup_reap_undispatched(me_inst.hostname) + wfj.refresh_from_db() + assert wfj.status == 'running', 'WorkflowJob must not be reaped by startup safety net' + + +@pytest.mark.django_db +def test_process_startup_jobs_skips_workflow_jobs(me_inst, settings): + """_process_startup_jobs() must not reap WorkflowJobs on controller restart. + + WorkflowJob has work_unit_id=None but is not a receptor-dispatched job; reaping it + on startup would immediately fail running workflows. + """ + settings.HADR_JOB_ADOPTION_TIMEOUT = 3600 + wfj = WorkflowJob.objects.create(status='running', controller_node=me_inst.hostname) + _process_startup_jobs(me_inst) + wfj.refresh_from_db() + assert wfj.status == 'running', 'WorkflowJob must not be reaped by startup job loop' + + +@pytest.mark.django_db +def test_configure_runner_callback_populates_host_map(me_inst): + """RunnerCallback.configure_for_job populates host_map from inventory hosts for adoption.""" + from awx.main.tasks.callback import RunnerCallback + from awx.main.models import Organization, Inventory, Host + + org = Organization.objects.first() + inv = Inventory.objects.create(name='cb-hostmap-inv', organization=org) + host = Host.objects.create(name='myhost', inventory=inv) + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-hm', inventory=inv) + + cb = RunnerCallback(model=Job) + cb.configure_for_job(job) + + assert cb.host_map.get('myhost') == host.id + + +@pytest.mark.django_db +def test_compute_adoption_dedup_no_events(me_inst): + """_compute_adoption_dedup returns (0, set()) when job has no events in DB.""" + from awx.main.tasks.receptor import _compute_adoption_dedup + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-1') + threshold, collision_zone = _compute_adoption_dedup(job) + assert threshold == 0 + assert collision_zone == set() + + +@pytest.mark.django_db +def test_compute_adoption_dedup_contiguous_events(me_inst): + """_compute_adoption_dedup returns (max_counter, empty set) for contiguous events.""" + from awx.main.tasks.receptor import _compute_adoption_dedup + from awx.main.models import JobEvent + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-1') + for ctr in [1, 2, 3, 4, 5]: + JobEvent.objects.create(job=job, counter=ctr, event='runner_on_ok') + + threshold, collision_zone = _compute_adoption_dedup(job) + assert threshold == 5 + assert collision_zone == set() + + +@pytest.mark.django_db +def test_compute_adoption_dedup_gap_produces_collision_zone(me_inst): + """_compute_adoption_dedup finds gap and puts above-gap events in collision_zone.""" + from awx.main.tasks.receptor import _compute_adoption_dedup + from awx.main.models import JobEvent + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-1') + # Events 1-3 contiguous, then gap at 4, then 5 and 6 committed out of order + for ctr in [1, 2, 3, 5, 6]: + JobEvent.objects.create(job=job, counter=ctr, event='runner_on_ok') + + threshold, collision_zone = _compute_adoption_dedup(job) + assert threshold == 3 + assert collision_zone == {5, 6} + + +@pytest.mark.django_db +def test_adopt_job_async_exception_is_swallowed(me_inst, settings): + """adopt_job_async swallows exceptions from reattach_to_work_unit and does not re-raise.""" + from awx.main.tasks.system import adopt_job_async + + settings.HADR_JOB_ADOPTION_TIMEOUT = 3600 + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-err') + with patch('awx.main.tasks.system.get_receptor_ctl'), patch('awx.main.tasks.system.reattach_to_work_unit', side_effect=RuntimeError('network failure')): + adopt_job_async(job.id) # must not raise + + +@pytest.mark.django_db +def test_adoption_counter_skip_dedup(): + """RunnerCallback.event_handler hybrid dedup: O(1) threshold + small collision-zone set. + + safe_threshold skips the contiguous prefix with a single integer comparison. + persisted_counters (collision_zone) handles out-of-order committed events above + the threshold — bounded by worker concurrency, not total event count. + """ + from awx.main.tasks.callback import RunnerCallback + + cb = RunnerCallback() + # Simulate: events 1-8 committed contiguously (threshold=8), + # events 9,10 committed out-of-order (collision zone) + cb.dedup_threshold = 8 + cb.persisted_counters = {9, 10} + + dispatched = [] + cb.dispatcher = MagicMock() + cb.dispatcher.dispatch.side_effect = dispatched.append + + from collections import deque + + cb.instance = MagicMock() + cb.instance.event_class.WRAPUP_EVENT = 'playbook_on_stats' + cb.event_data_key = 'job_id' + cb.job_created = None + cb.parent_workflow_job_id = None + cb.host_map = {} + cb.recent_event_timings = deque(maxlen=100) + + # below threshold — skipped by O(1) integer check + cb.event_handler({'event': 'runner_on_ok', 'counter': 5, 'job_id': 1}) + assert len(dispatched) == 0, 'counter <= threshold should be skipped' + + # at threshold — also skipped + cb.event_handler({'event': 'runner_on_ok', 'counter': 8, 'job_id': 1}) + assert len(dispatched) == 0, 'counter == threshold should be skipped' + + # in collision zone — skipped by set membership + cb.event_handler({'event': 'runner_on_ok', 'counter': 9, 'job_id': 1}) + assert len(dispatched) == 0, 'counter in collision_zone should be skipped' + + # None threshold disables dedup entirely (normal job path, zero overhead) + cb.dedup_threshold = None + cb.persisted_counters = None + cb.event_handler({'event': 'runner_on_ok', 'counter': 1, 'job_id': 1}) + # no assertion — just verify no exception + + +# ── reattach_to_work_unit branch coverage ──────────────────────────────────── + + +@pytest.mark.django_db +def test_reattach_receptor_command_fails(me_inst): + """receptor_ctl.simple_command raising returns False without touching the job.""" + from awx.main.tasks.receptor import reattach_to_work_unit + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-err') + ctl = MagicMock() + ctl.simple_command.side_effect = Exception('connection refused') + + result = reattach_to_work_unit(job, ctl) + + assert result is False + job.refresh_from_db() + assert job.status == 'running' + + +@pytest.mark.django_db +def test_reattach_exit_code_from_detail(me_inst): + """Exit code is parsed from the Detail string when ExitCode is absent.""" + from awx.main.tasks.receptor import reattach_to_work_unit + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-detail') + ctl = MagicMock() + ctl.simple_command.return_value = {'StateName': 'Failed', 'Detail': 'exit status 2'} + + with ( + patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls, + patch('awx.main.tasks.callback.RunnerCallback'), + ): + mock_job_cls.return_value = MagicMock() + reattach_to_work_unit(job, ctl) + + job.refresh_from_db() + assert job.status == 'failed' + + +@pytest.mark.django_db +def test_reattach_exit_code_fallback_succeeded(me_inst): + """_process_phase returning status='successful' → job finalized as successful.""" + from awx.main.tasks.receptor import reattach_to_work_unit + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-fb-ok') + ctl = MagicMock() + ctl.simple_command.return_value = {'StateName': 'Succeeded', 'Detail': 'not-a-number'} + + with ( + patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls, + patch('awx.main.tasks.callback.RunnerCallback'), + ): + mock_instance = MagicMock() + mock_instance._process_phase.return_value = MagicMock(status='successful', rc=0) + mock_job_cls.return_value = mock_instance + reattach_to_work_unit(job, ctl) + + job.refresh_from_db() + assert job.status == 'successful' + + +@pytest.mark.django_db +def test_reattach_exit_code_fallback_failed(me_inst): + """Unparseable Detail with Failed state → exit_code=1 → failed.""" + from awx.main.tasks.receptor import reattach_to_work_unit + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-fb-fail') + ctl = MagicMock() + ctl.simple_command.return_value = {'StateName': 'Failed', 'Detail': ''} + + with ( + patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls, + patch('awx.main.tasks.callback.RunnerCallback'), + ): + mock_job_cls.return_value = MagicMock() + reattach_to_work_unit(job, ctl) + + job.refresh_from_db() + assert job.status == 'failed' + + +@pytest.mark.django_db +def test_reattach_process_phase_raises(me_inst): + """When _process_phase raises, the job is still finalized via the pre-fetched exit_code.""" + from awx.main.tasks.receptor import reattach_to_work_unit + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-raise') + ctl = MagicMock() + ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0} + + with ( + patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls, + patch('awx.main.tasks.callback.RunnerCallback'), + ): + mock_instance = MagicMock() + mock_instance._process_phase.side_effect = RuntimeError('boom') + mock_job_cls.return_value = mock_instance + result = reattach_to_work_unit(job, ctl) + + assert result is True + job.refresh_from_db() + assert job.status == 'successful' + + +@pytest.mark.django_db +def test_reattach_job_already_finalized(me_inst): + """finished_callback finalizing the job during process phase is not overwritten.""" + from awx.main.tasks.receptor import reattach_to_work_unit + + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-already') + ctl = MagicMock() + ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0} + + def _finalize_in_db(*args, **kwargs): + Job.objects.filter(pk=job.pk).update(status='successful') + + with ( + patch('awx.main.tasks.receptor.AWXReceptorJob') as mock_job_cls, + patch('awx.main.tasks.callback.RunnerCallback'), + ): + mock_instance = MagicMock() + mock_instance._process_phase.side_effect = _finalize_in_db + mock_job_cls.return_value = mock_instance + reattach_to_work_unit(job, ctl) + + job.refresh_from_db() + assert job.status == 'successful' + + +@pytest.mark.django_db +def test_adopt_job_async_calls_reattach(me_inst, settings): + """adopt_job_async creates its own receptor_ctl and calls reattach_to_work_unit.""" + from awx.main.tasks.system import adopt_job_async + + settings.HADR_JOB_ADOPTION_TIMEOUT = 3600 + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-async') + + with patch('awx.main.tasks.system.get_receptor_ctl') as mock_ctl_factory, patch('awx.main.tasks.system.reattach_to_work_unit') as mock_reattach: + adopt_job_async(job.id) + + mock_ctl_factory.assert_called_once() + mock_reattach.assert_called_once_with(job, mock_ctl_factory.return_value) + + +@pytest.mark.django_db +def test_adopt_job_async_skips_already_finalized(me_inst, settings): + """adopt_job_async is a no-op if the job is no longer running.""" + from awx.main.tasks.system import adopt_job_async + + settings.HADR_JOB_ADOPTION_TIMEOUT = 3600 + job = Job.objects.create(controller_node=me_inst.hostname, status='successful', work_unit_id='unit-done') + + with patch('awx.main.tasks.system.reattach_to_work_unit') as mock_reattach: + adopt_job_async(job.id) + + mock_reattach.assert_not_called() + + +@pytest.mark.django_db +def test_adopt_job_async_reaps_on_timeout(me_inst, settings): + """adopt_job_async reaps a job orphaned longer than HADR_JOB_ADOPTION_TIMEOUT.""" + from awx.main.tasks.system import adopt_job_async + + settings.HADR_JOB_ADOPTION_TIMEOUT = 3600 + job = Job.objects.create( + controller_node=me_inst.hostname, + status='running', + work_unit_id='unit-old', + started=now() - timedelta(seconds=7200), + ) + + with patch('awx.main.tasks.system.reattach_to_work_unit') as mock_reattach: + adopt_job_async(job.id) + + mock_reattach.assert_not_called() + job.refresh_from_db() + assert job.status == 'failed' + + +@pytest.mark.django_db +def test_adopt_job_async_ctl_close_exception_is_swallowed(me_inst, settings): + """adopt_job_async does not propagate exceptions from receptor_ctl.close().""" + from awx.main.tasks.system import adopt_job_async + + settings.HADR_JOB_ADOPTION_TIMEOUT = 3600 + job = Job.objects.create(controller_node=me_inst.hostname, status='running', work_unit_id='unit-close-err') + + mock_ctl = MagicMock() + mock_ctl.close.side_effect = RuntimeError('socket already closed') + + with patch('awx.main.tasks.system.get_receptor_ctl', return_value=mock_ctl), patch('awx.main.tasks.system.reattach_to_work_unit'): + adopt_job_async(job.id) # must not raise + + +# ── _finalize_job_run coverage ────────────────────────────────────────────── + + +@pytest.mark.django_db +def test_finalize_job_run_with_extra_fields(me_inst): + """_finalize_job_run includes extra_fields in the update.""" + from awx.main.tasks.jobs import _finalize_job_run + from awx.main.tasks.callback import RunnerCallback + + job = Job.objects.create(controller_node=me_inst.hostname, status='running') + callback = RunnerCallback(model=Job) + callback.instance = job + callback.wrapup_event_dispatched = True + + _finalize_job_run(Job, job.pk, callback, 'successful', extra_fields={'elapsed': 100.5}) + + job.refresh_from_db() + assert job.status == 'successful' + assert job.elapsed == 100.5 + + +@pytest.mark.django_db +def test_finalize_job_run_update_model_returns_none(me_inst): + """_finalize_job_run returns None when update_model fails.""" + from awx.main.tasks.jobs import _finalize_job_run + from awx.main.tasks.callback import RunnerCallback + + callback = RunnerCallback(model=Job) + callback.wrapup_event_dispatched = True + + with patch('awx.main.tasks.jobs.update_model', return_value=None): + result = _finalize_job_run(Job, 99999, callback, 'successful') + + assert result is None + + +@pytest.mark.django_db +def test_finalize_job_run_with_blocked_jobs(me_inst): + """_finalize_job_run schedules task manager when job has blocked dependents.""" + from awx.main.tasks.jobs import _finalize_job_run + from awx.main.tasks.callback import RunnerCallback + + job = Job.objects.create(controller_node=me_inst.hostname, status='running') + callback = RunnerCallback(model=Job) + callback.instance = job + callback.wrapup_event_dispatched = True + + with patch('awx.main.tasks.jobs.ScheduleTaskManager') as mock_tm: + with patch('awx.main.tasks.jobs.update_model') as mock_update: + # Create a mock instance that has unifiedjob_blocked_jobs + mock_instance = MagicMock() + mock_instance.unifiedjob_blocked_jobs.exists.return_value = True + mock_update.return_value = mock_instance + _finalize_job_run(Job, job.pk, callback, 'successful') + mock_tm.return_value.schedule.assert_called() + + +@pytest.mark.django_db +def test_finalize_job_run_with_inventory_update(me_inst): + """_finalize_job_run schedules inventory computed fields update for jobs with inventory.""" + from awx.main.tasks.jobs import _finalize_job_run + from awx.main.tasks.callback import RunnerCallback + from awx.main.models import Organization, Inventory + + org = Organization.objects.first() + inv = Inventory.objects.create(name='test-inv', organization=org) + job = Job.objects.create(controller_node=me_inst.hostname, status='running', inventory=inv) + + callback = RunnerCallback(model=Job) + callback.instance = job + callback.wrapup_event_dispatched = True + + with patch('awx.main.tasks.jobs.update_inventory_computed_fields') as mock_delay: + _finalize_job_run(Job, job.pk, callback, 'successful') + mock_delay.delay.assert_called_once_with(inv.id) + + +@pytest.mark.django_db +def test_finalize_job_run_inventory_update_exception_logged(me_inst): + """_finalize_job_run logs exception from inventory computed fields update.""" + from awx.main.tasks.jobs import _finalize_job_run + from awx.main.tasks.callback import RunnerCallback + from awx.main.models import Organization, Inventory + + org = Organization.objects.first() + inv = Inventory.objects.create(name='test-inv-err', organization=org) + job = Job.objects.create(controller_node=me_inst.hostname, status='running', inventory=inv) + + callback = RunnerCallback(model=Job) + callback.instance = job + callback.wrapup_event_dispatched = True + + with patch('awx.main.tasks.jobs.update_inventory_computed_fields') as mock_delay: + mock_delay.delay.side_effect = RuntimeError('celery error') + with patch('awx.main.tasks.jobs.logger') as mock_logger: + _finalize_job_run(Job, job.pk, callback, 'successful') + mock_logger.exception.assert_called() diff --git a/awx/main/tests/functional/test_jobs.py b/awx/main/tests/functional/test_jobs.py index a9d4278480..aa2f996b5a 100644 --- a/awx/main/tests/functional/test_jobs.py +++ b/awx/main/tests/functional/test_jobs.py @@ -199,7 +199,6 @@ class TestAnsibleFactsSave: @pytest.mark.django_db def test_update_hosts_resolved_deadlock(inventory, mocker): - hosts = [Host.objects.create(inventory=inventory, name=f'foo{i}') for i in range(3)] # Set ansible_facts for each host diff --git a/awx/main/tests/unit/tasks/test_jobs.py b/awx/main/tests/unit/tasks/test_jobs.py index 0fc0fa98b6..0763d44bb6 100644 --- a/awx/main/tests/unit/tasks/test_jobs.py +++ b/awx/main/tests/unit/tasks/test_jobs.py @@ -661,3 +661,209 @@ class TestRunInventoryUpdatePopulateWorkloadIdentityTokens: # The instance's get_cloud_credential should now return the same object with context assert task.instance.get_cloud_credential() is cloud_cred + + def test_no_cloud_credential_when_none_returned_from_instance(self): + """When instance.get_cloud_credential returns None, populate_workload_identity_tokens is called with None.""" + task = jobs.RunInventoryUpdate() + task.instance = mock.MagicMock() + task.instance.get_cloud_credential.return_value = None + task._credentials = [] + + with mock.patch.object(jobs.BaseTask, 'populate_workload_identity_tokens') as mock_super: + task.populate_workload_identity_tokens() + + mock_super.assert_called_once_with(additional_credentials=None) + + +def test_work_unit_released_on_quota_exceeded(): + """ + Work unit is released on quota exceeded (when res is None). + + Scenario: Job hits quota limit → _handle_work_error returns None → + BaseTask.run() early-returns. Work unit MUST be released. + """ + receptor_ctl = mock.MagicMock() + receptor_job = mock.MagicMock() + receptor_job.unit_id = "test-unit-123" + receptor_job.receptor_ctl = receptor_ctl + + # When res is None (quota exceeded case) + res = None + + # This is the fixed code path from jobs.py lines 810-818 + if not res: + if receptor_job and getattr(receptor_job, 'receptor_ctl', None): + try: + receptor_job._receptor_release_work(receptor_job.receptor_ctl, 'error') + except Exception: + pass + + # Verify release was called + receptor_job._receptor_release_work.assert_called_once_with(receptor_ctl, 'error') + + +def test_work_unit_not_released_if_no_receptor_ctl(): + """ + Edge case: receptor_ctl not set (transmit phase failed before unit submission). + + Verify graceful handling when receptor_ctl is None. + """ + receptor_job = mock.MagicMock(spec=['unit_id', '_receptor_release_work']) + receptor_job.receptor_ctl = None # Not set + + res = None + + # Code should not crash + if not res: + if receptor_job and getattr(receptor_job, 'receptor_ctl', None): + receptor_job._receptor_release_work(receptor_job.receptor_ctl, 'error') + # else: No release needed, unit was never submitted + + # Verify _receptor_release_work was NOT called + assert receptor_job._receptor_release_work.call_count == 0 + + +@mock.patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@mock.patch('awx.main.tasks.jobs.ScheduleTaskManager') +@mock.patch('awx.main.tasks.jobs.events_processed_hook') +@mock.patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_with_host_status_counts(mock_update, mock_hook, mock_task_mgr, mock_workflow_mgr): + """events_processed_hook is called when host_status_counts is not None.""" + instance = mock.MagicMock() + instance.host_status_counts = {'successful': 5} # Not None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = mock.MagicMock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + jobs._finalize_job_run(mock.MagicMock, pk=1, runner_callback=cb, status='successful') + + mock_hook.assert_called_once_with(instance) + + +@mock.patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@mock.patch('awx.main.tasks.jobs.ScheduleTaskManager') +@mock.patch('awx.main.tasks.jobs.events_processed_hook') +@mock.patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_with_wrapup_not_dispatched(mock_update, mock_hook, mock_task_mgr, mock_workflow_mgr): + """events_processed_hook is called when wrapup_event_dispatched is False.""" + instance = mock.MagicMock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = mock.MagicMock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = False # Not dispatched + + jobs._finalize_job_run(mock.MagicMock, pk=1, runner_callback=cb, status='successful') + + mock_hook.assert_called_once_with(instance) + + +@mock.patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@mock.patch('awx.main.tasks.jobs.ScheduleTaskManager') +@mock.patch('awx.main.tasks.jobs.events_processed_hook') +@mock.patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_calls_log_lifecycle(mock_update, mock_hook, mock_task_mgr, mock_workflow_mgr): + """instance.log_lifecycle('finalize_run') is called.""" + instance = mock.MagicMock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = mock.MagicMock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + jobs._finalize_job_run(mock.MagicMock, pk=1, runner_callback=cb, status='successful') + + instance.log_lifecycle.assert_called_once_with('finalize_run') + + +@mock.patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@mock.patch('awx.main.tasks.jobs.ScheduleTaskManager') +@mock.patch('awx.main.tasks.jobs.events_processed_hook') +@mock.patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_calls_websocket_emit(mock_update, mock_hook, mock_task_mgr, mock_workflow_mgr): + """instance.websocket_emit_status is called with the final status.""" + instance = mock.MagicMock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = mock.MagicMock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + jobs._finalize_job_run(mock.MagicMock, pk=1, runner_callback=cb, status='failed') + + instance.websocket_emit_status.assert_called_once_with('failed') + + +@mock.patch('awx.main.tasks.jobs.fcntl.lockf') +@mock.patch('awx.main.tasks.jobs.os.open') +@mock.patch('awx.main.tasks.jobs.os.path.exists') +def test_acquire_lock_shared(mock_exists, mock_os_open, mock_lockf): + """acquire_lock with exclusive=False uses LOCK_SH.""" + import fcntl + from awx.main.tasks.jobs import BaseTask + + # Setup mocks + mock_exists.return_value = True + mock_os_open.return_value = 5 + task = BaseTask() + task.instance = mock.MagicMock() + task.instance.refresh_from_db = mock.MagicMock() + task.instance.cancel_flag = False + + project = mock.MagicMock() + project.get_lock_file.return_value = '/tmp/project.lock' + + # Mock signal_callback to return False (not cancelled) + with mock.patch('awx.main.tasks.jobs.signal_callback', return_value=False): + task.acquire_lock(project, unified_job_id=1, exclusive=False) + + # Verify LOCK_SH (not LOCK_EX) was used + call_args = mock_lockf.call_args + lock_flags = call_args[0][1] + assert lock_flags & fcntl.LOCK_SH # Should have LOCK_SH bit + + +@mock.patch('awx.main.tasks.jobs.fcntl.lockf') +@mock.patch('awx.main.tasks.jobs.os.open') +@mock.patch('awx.main.tasks.jobs.os.path.exists') +def test_acquire_lock_exclusive(mock_exists, mock_os_open, mock_lockf): + """acquire_lock with exclusive=True uses LOCK_EX.""" + import fcntl + from awx.main.tasks.jobs import BaseTask + + # Setup mocks + mock_exists.return_value = True + mock_os_open.return_value = 5 + task = BaseTask() + task.instance = mock.MagicMock() + task.instance.refresh_from_db = mock.MagicMock() + task.instance.cancel_flag = False + + project = mock.MagicMock() + project.get_lock_file.return_value = '/tmp/project.lock' + + # Mock signal_callback to return False (not cancelled) + with mock.patch('awx.main.tasks.jobs.signal_callback', return_value=False): + task.acquire_lock(project, unified_job_id=1, exclusive=True) + + # Verify LOCK_EX was used + call_args = mock_lockf.call_args + lock_flags = call_args[0][1] + assert lock_flags & fcntl.LOCK_EX # Should have LOCK_EX bit diff --git a/awx/main/tests/unit/tasks/test_receptor_adoption.py b/awx/main/tests/unit/tasks/test_receptor_adoption.py new file mode 100644 index 0000000000..2188c3d41b --- /dev/null +++ b/awx/main/tests/unit/tasks/test_receptor_adoption.py @@ -0,0 +1,1416 @@ +"""Unit tests for receptor job adoption (same-controller restart reattach). + +Covers: + - reattach_to_work_unit: all branch paths + - AWXReceptorJob._process_phase / _handle_work_error: all branch paths + - receptor_config_exists + - _get_or_create_private_data_dir + - should_update_config FileNotFoundError path +""" + +import json +import socket +from collections import namedtuple +from unittest.mock import MagicMock, Mock, patch + +import pytest + +from awx.main.models import Job +from awx.main.tasks.callback import RunnerCallback +from awx.main.tasks.jobs import _finalize_job_run +from awx.main.tasks.receptor import ( + AWXReceptorJob, + _AdoptionTask, + _compute_adoption_dedup, + _finalize_adopted_job, + _get_adoption_exit_code, + _get_or_create_private_data_dir, + adopt_remote_work, + receptor_config_exists, + reattach_to_work_unit, + should_update_config, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_Result = namedtuple('result', ['status', 'rc']) + + +def _make_receptor_job(unit_id='unit-1', extra_update_fields=None, event_ct=0): + """Return a minimal AWXReceptorJob built without touching Django models.""" + rj = AWXReceptorJob.__new__(AWXReceptorJob) + task = Mock() + task.instance.is_container_group_task = False + task.instance.execution_node = 'remote-node' + task.instance.controller_node = 'controller-node' + task.instance.log_format = 'job 1' + task.instance.pk = 1 + task.instance.id = 1 + task.instance.work_unit_id = unit_id + task.runner_callback.extra_update_fields = extra_update_fields if extra_update_fields is not None else {} + task.runner_callback.event_ct = event_ct + rj.task = task + rj.unit_id = unit_id + rj.runner_params = {'private_data_dir': '/tmp/test'} + return rj + + +def _make_receptor_ctl(state='Succeeded', exit_code=None, detail='', stdout_size=0): + """Return a mock ReceptorControl with configurable work-status response.""" + ctl = Mock() + status = {'StateName': state, 'Detail': detail, 'StdoutSize': stdout_size} + if exit_code is not None: + status['ExitCode'] = exit_code + ctl.simple_command.return_value = status + sock_mock = Mock(spec=socket.socket) + file_mock = MagicMock() + file_mock.readlines.return_value = [b'some output'] + ctl.get_work_results.return_value = (sock_mock, file_mock) + return ctl + + +# --------------------------------------------------------------------------- +# receptor_config_exists +# --------------------------------------------------------------------------- + + +def test_receptor_config_exists_true(): + with patch('awx.main.tasks.receptor.os.path.exists', return_value=True): + assert receptor_config_exists() is True + + +def test_receptor_config_exists_false(): + with patch('awx.main.tasks.receptor.os.path.exists', return_value=False): + assert receptor_config_exists() is False + + +# --------------------------------------------------------------------------- +# _get_or_create_private_data_dir +# --------------------------------------------------------------------------- + + +def test_get_or_create_private_data_dir(tmp_path): + job = Mock() + job.pk = 42 + with patch('awx.main.tasks.receptor.settings') as s: + s.AWX_ISOLATION_BASE_PATH = str(tmp_path) + result = _get_or_create_private_data_dir(job) + import os + + assert os.path.isdir(result) + assert 'adoption_' in result + os.rmdir(result) + + +# --------------------------------------------------------------------------- +# should_update_config — FileNotFoundError path (new in this PR) +# --------------------------------------------------------------------------- + + +def test_should_update_config_file_not_found(): + with patch('awx.main.tasks.receptor.read_receptor_config', side_effect=FileNotFoundError): + assert should_update_config([]) is True + + +# --------------------------------------------------------------------------- +# AWXReceptorJob._process_phase — success path +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.receptor.connections') +@patch('awx.main.tasks.receptor.signal_callback', return_value=False) +def test_process_phase_success(mock_signal, mock_connections): + rj = _make_receptor_job() + expected_res = _Result(status='successful', rc=0) + rj.processor = Mock(return_value=expected_res) + ctl = _make_receptor_ctl() + + res = rj._process_phase(ctl) + + assert res.status == 'successful' + mock_connections.close_all.assert_called_once() + + +# --------------------------------------------------------------------------- +# AWXReceptorJob._process_phase — SignalExit path +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.receptor.connections') +@patch('awx.main.tasks.receptor.signal_callback', return_value=True) +def test_process_phase_signal_exit(mock_signal, mock_connections): + rj = _make_receptor_job() + rj.processor = Mock() + ctl = _make_receptor_ctl() + + res = rj._process_phase(ctl) + + assert res.status == 'canceled' + ctl.simple_command.assert_any_call('work cancel unit-1') + + +# --------------------------------------------------------------------------- +# _handle_work_error — result_traceback already recorded → return res early +# --------------------------------------------------------------------------- + + +def test_handle_work_error_result_traceback_present(): + rj = _make_receptor_job(extra_update_fields={'result_traceback': 'boom'}) + ctl = _make_receptor_ctl() + err_res = _Result(status='error', rc=1) + + result = rj._handle_work_error(ctl, err_res) + + assert result is err_res + ctl.simple_command.assert_not_called() + + +# --------------------------------------------------------------------------- +# _handle_work_error — status command raises exception +# --------------------------------------------------------------------------- + + +def test_handle_work_error_status_command_raises(): + rj = _make_receptor_job() + ctl = Mock() + ctl.simple_command.side_effect = Exception('network error') + err_res = _Result(status='error', rc=1) + + # should not raise; falls through and returns res + result = rj._handle_work_error(ctl, err_res) + assert result is err_res + + +# --------------------------------------------------------------------------- +# _handle_work_error — 'exceeded quota' in detail +# --------------------------------------------------------------------------- + + +def test_handle_work_error_exceeded_quota(): + rj = _make_receptor_job() + ctl = _make_receptor_ctl(detail='exceeded quota for namespace') + err_res = _Result(status='error', rc=1) + + result = rj._handle_work_error(ctl, err_res) + + assert result is None + rj.task.update_model.assert_called_once_with(1, status='pending') + + +# --------------------------------------------------------------------------- +# _handle_work_error — Failed state with no events → reads receptor stdout +# --------------------------------------------------------------------------- + + +def test_handle_work_error_reads_receptor_output_on_failed(): + rj = _make_receptor_job(event_ct=0) + ctl = _make_receptor_ctl(state='Failed', stdout_size=2000) + err_res = _Result(status='error', rc=1) + + rj._handle_work_error(ctl, err_res) + + rj.task.runner_callback.delay_update.assert_called_once() + call_kwargs = rj.task.runner_callback.delay_update.call_args[1] + assert 'Worker output' in call_kwargs.get('result_traceback', '') + + +# --------------------------------------------------------------------------- +# _handle_work_error — detail present, no receptor output → delay_update with detail +# --------------------------------------------------------------------------- + + +def test_handle_work_error_detail_only(): + rj = _make_receptor_job(event_ct=1) # event_ct > 0, so no stdout fetch + ctl = _make_receptor_ctl(state='Succeeded', detail='some error detail') + err_res = _Result(status='error', rc=1) + + rj._handle_work_error(ctl, err_res) + + rj.task.runner_callback.delay_update.assert_called_once() + call_kwargs = rj.task.runner_callback.delay_update.call_args[1] + assert 'Receptor detail' in call_kwargs.get('result_traceback', '') + + +# --------------------------------------------------------------------------- +# _handle_work_error — no detail, no output → logs warning, returns res +# --------------------------------------------------------------------------- + + +def test_handle_work_error_no_detail_no_output(): + rj = _make_receptor_job(event_ct=1) + ctl = _make_receptor_ctl(state='Succeeded', detail='') + err_res = _Result(status='error', rc=1) + + result = rj._handle_work_error(ctl, err_res) + + assert result is err_res + rj.task.runner_callback.delay_update.assert_not_called() + + +# --------------------------------------------------------------------------- +# _handle_work_error — get_work_results raises → RuntimeError propagated +# --------------------------------------------------------------------------- + + +def test_handle_work_error_get_results_raises(): + rj = _make_receptor_job(event_ct=0) + ctl = _make_receptor_ctl(state='Failed', stdout_size=500) + ctl.get_work_results.side_effect = Exception('socket error') + err_res = _Result(status='error', rc=1) + + with pytest.raises(RuntimeError): + rj._handle_work_error(ctl, err_res) + + +# --------------------------------------------------------------------------- +# reattach_to_work_unit — callback.job_created is set from job.created +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_sets_job_created_on_callback(mock_rmtree, mock_pdd, mock_release, mock_process, mock_dedup, mock_finalize): + """callback.job_created must be set so events are stored with the correct timestamp.""" + job = Mock() + job.id = 1 + job.work_unit_id = 'unit-1' + job.created = '2026-01-01T00:00:00Z' + job.spawned_by_workflow = False + job.status = 'successful' + ctl = Mock() + ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0, 'Detail': ''} + + from awx.main.tasks.callback import RunnerCallback as RealCallback + + created_callbacks = [] + original_cb_init = RealCallback.__init__ + + def capturing_cb_init(self, model=None): + original_cb_init(self, model) + created_callbacks.append(self) + + with patch.object(RealCallback, '__init__', capturing_cb_init): + reattach_to_work_unit(job, ctl) + + assert len(created_callbacks) == 1 + assert created_callbacks[0].job_created == str(job.created) + + +# --------------------------------------------------------------------------- +# reattach_to_work_unit — _receptor_release_work called after success and failure +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_releases_work_unit_on_success(mock_rmtree, mock_pdd, mock_release, mock_process, mock_dedup, mock_finalize): + """_receptor_release_work must be called after _finalize_adopted_job.""" + job = Mock() + job.id = 1 + job.work_unit_id = 'unit-1' + job.spawned_by_workflow = False + job.status = 'successful' + job.started = None + ctl = Mock() + ctl.simple_command.return_value = {'StateName': ''} + mock_process.return_value = Mock(status='successful', rc=0) + + reattach_to_work_unit(job, ctl) + + mock_release.assert_called_once() + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase', side_effect=RuntimeError('network failure')) +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_releases_work_unit_on_failure(mock_rmtree, mock_pdd, mock_release, mock_process, mock_dedup, mock_finalize): + """_receptor_release_work must run even when _process_phase raises.""" + job = Mock() + job.id = 1 + job.work_unit_id = 'unit-1' + job.spawned_by_workflow = False + job.started = None + job.status = 'running' + ctl = Mock() + ctl.simple_command.side_effect = [ + {'StateName': ''}, # initial state check (unknown → _process_phase) + {'StateName': 'Succeeded', 'ExitCode': 0, 'Detail': ''}, # fallback re-check + ] + + result = reattach_to_work_unit(job, ctl) + + assert result is True # finalized via exit_code path + mock_release.assert_called_once() + mock_finalize.assert_called_once() # _finalize_job_run was invoked + + +# --------------------------------------------------------------------------- +# reattach_to_work_unit — spawned_by_workflow=True → parent_workflow_job_id set +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_sets_parent_workflow_job_id_when_workflow_child(mock_rmtree, mock_pdd, mock_release, mock_process, mock_dedup, mock_finalize): + """callback.parent_workflow_job_id must be set for workflow-child jobs so events are + correctly associated with their parent workflow in the event stream.""" + job = Mock() + job.id = 1 + job.work_unit_id = 'unit-1' + job.created = '2026-01-01T00:00:00Z' + job.spawned_by_workflow = True + job.get_workflow_job.return_value.id = 999 + job.status = 'successful' + ctl = Mock() + ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0, 'Detail': ''} + + from awx.main.tasks.callback import RunnerCallback as RealCallback + + created_callbacks = [] + original_cb_init = RealCallback.__init__ + + def capturing_cb_init(self, model=None): + original_cb_init(self, model) + created_callbacks.append(self) + + with patch.object(RealCallback, '__init__', capturing_cb_init): + reattach_to_work_unit(job, ctl) + + assert len(created_callbacks) == 1 + assert created_callbacks[0].parent_workflow_job_id == 999 + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_workflow_job_lookup_exception_swallowed(mock_rmtree, mock_pdd, mock_release, mock_process, mock_dedup, mock_finalize): + """If get_workflow_job() raises, adoption must still complete (bare except: pass).""" + job = Mock() + job.id = 1 + job.work_unit_id = 'unit-1' + job.created = '2026-01-01T00:00:00Z' + job.spawned_by_workflow = True + job.get_workflow_job.side_effect = Exception('workflow lookup failed') + job.status = 'successful' + ctl = Mock() + ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0, 'Detail': ''} + + result = reattach_to_work_unit(job, ctl) + + assert result is True # exception swallowed, adoption completes + + +# --------------------------------------------------------------------------- +# _get_adoption_exit_code — all three branches +# --------------------------------------------------------------------------- + + +def test_get_adoption_exit_code_from_key(): + assert _get_adoption_exit_code({'ExitCode': 42, 'Detail': 'ignored'}, 'Succeeded') == 42 + + +def test_get_adoption_exit_code_from_detail_string(): + assert _get_adoption_exit_code({'Detail': 'exit status 1'}, 'Failed') == 1 + + +def test_get_adoption_exit_code_fallback_succeeded(): + assert _get_adoption_exit_code({'Detail': 'no int here!'}, 'Succeeded') == 0 + + +def test_get_adoption_exit_code_fallback_failed(): + assert _get_adoption_exit_code({}, 'Failed') == 1 + + +# --------------------------------------------------------------------------- +# _AdoptionTask — method coverage +# --------------------------------------------------------------------------- + + +def test_adoption_task_build_ee_params_returns_empty_dict(): + task = _AdoptionTask(Mock(), Mock()) + assert task.build_execution_environment_params(None, None) == {} + + +def test_adoption_task_update_model_is_noop(): + task = _AdoptionTask(Mock(), Mock()) + task.update_model(1, status='successful', result_traceback='boom') # must not raise + + +# --------------------------------------------------------------------------- +# _configure_runner_callback — shared initialization for normal + adoption paths +# --------------------------------------------------------------------------- + + +def test_configure_runner_callback_sets_common_fields(): + from awx.main.tasks.callback import RunnerCallback + + cb = RunnerCallback(model=None) + instance = Mock() + instance.created = '2026-01-01T00:00:00Z' + instance.spawned_by_workflow = False + + cb.configure_for_job(instance, safe_env={'KEY': 'val'}, dedup_threshold=5, persisted_counters={6, 7}) + + assert cb.instance is instance + assert cb.job_created == str(instance.created) + assert cb.safe_env == {'KEY': 'val'} + assert cb.dedup_threshold == 5 + assert cb.persisted_counters == {6, 7} + + +def test_configure_runner_callback_safe_env_defaults_to_empty_dict(): + from awx.main.tasks.callback import RunnerCallback + + cb = RunnerCallback(model=None) + instance = Mock() + instance.created = '2026-01-01' + instance.spawned_by_workflow = False + + cb.configure_for_job(instance) + + assert cb.safe_env == {} + assert cb.dedup_threshold is None + assert cb.persisted_counters is None + + +def test_configure_runner_callback_sets_parent_workflow_job_id(): + from awx.main.tasks.callback import RunnerCallback + + cb = RunnerCallback(model=None) + instance = Mock() + instance.created = '2026-01-01' + instance.spawned_by_workflow = True + instance.get_workflow_job.return_value.id = 42 + + cb.configure_for_job(instance) + + assert cb.parent_workflow_job_id == 42 + + +def test_configure_runner_callback_swallows_workflow_lookup_error(): + from awx.main.tasks.callback import RunnerCallback + + cb = RunnerCallback(model=None) + instance = Mock() + instance.created = '2026-01-01' + instance.spawned_by_workflow = True + instance.get_workflow_job.side_effect = Exception('lookup failed') + + cb.configure_for_job(instance) # must not raise + assert cb.parent_workflow_job_id is None + + +# --------------------------------------------------------------------------- +# _build_adoption_callback — spawned_by_workflow False path +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_non_workflow_job_no_parent_id(mock_rmtree, mock_pdd, mock_release, mock_process, mock_dedup, mock_finalize): + """callback.parent_workflow_job_id is NOT set for non-workflow jobs.""" + job = Mock() + job.id = 1 + job.work_unit_id = 'unit-1' + job.created = '2026-01-01T00:00:00Z' + job.spawned_by_workflow = False + job.status = 'successful' + ctl = Mock() + ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0, 'Detail': ''} + + from awx.main.tasks.callback import RunnerCallback as RealCallback + + created_callbacks = [] + original_cb_init = RealCallback.__init__ + + def capturing_cb_init(self, model=None): + original_cb_init(self, model) + created_callbacks.append(self) + + with patch.object(RealCallback, '__init__', capturing_cb_init): + reattach_to_work_unit(job, ctl) + + assert created_callbacks[0].parent_workflow_job_id is None + + +# --------------------------------------------------------------------------- +# reattach_to_work_unit — streams immediately even for still-running units +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_running_defers_without_streaming_variant(mock_rmtree, mock_pdd, mock_release, mock_process, mock_dedup, mock_finalize): + """Running state returns False — stdout forwarding is not available for adopted units. + + PR#1564 work adopt starts a status-monitoring goroutine but does not proxy the EE's + stdout bytes to the adopting node. Calling _process_phase → get_work_results would + block forever on the empty local stdout file. Return False so the heartbeat retries. + """ + job = Mock() + job.id = 1 + job.work_unit_id = 'unit-running' + job.created = '2026-01-01T00:00:00Z' + job.spawned_by_workflow = False + job.status = 'running' + job.started = None + ctl = Mock() + ctl.simple_command.return_value = {'StateName': 'Running'} + + result = reattach_to_work_unit(job, ctl) + + mock_process.assert_not_called() + assert result is False + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_process_phase_failure_falls_back_to_work_status(mock_rmtree, mock_pdd, mock_release, mock_process, mock_dedup, mock_finalize): + """When _process_phase raises, exit code falls back to work status re-check. + + If the re-check also fails, the exception is swallowed and the job is finalized + with exit_code=1 (failed). This path only triggers when _process_phase itself + raises (process_phase_failed=True), not when the job streams normally. + """ + job = Mock() + job.id = 1 + job.work_unit_id = 'unit-recheck-err' + job.created = '2026-01-01T00:00:00Z' + job.spawned_by_workflow = False + job.status = 'running' + job.started = None + ctl = Mock() + # Initial check: Running (falls through to _process_phase); re-check after raise also raises + ctl.simple_command.side_effect = [ + {'StateName': ''}, # initial state check (unknown → _process_phase) + RuntimeError('socket closed'), + ] + mock_process.side_effect = RuntimeError('process phase failed') + + result = reattach_to_work_unit(job, ctl) # must not raise + + # Returns True (function completed without re-raising) + assert result is True + + +# --------------------------------------------------------------------------- +# reattach_to_work_unit — state guards (Pending / terminal / Running) +# --------------------------------------------------------------------------- + + +def test_reattach_pending_defers_without_streaming(): + """Pending state returns False immediately — avoids the infinite get_work_results block. + + When work adopt creates a cross-controller adoption unit it starts in Pending while + the receptor mesh connects. get_work_results would block forever because the adopted + unit's stdout pipe is not yet (and may never be) forwarded. We return False so + adopt_job_async exits cleanly and _process_running_jobs re-queues it next heartbeat. + """ + job = Mock() + job.id = 42 + job.work_unit_id = 'unit-pending' + ctl = Mock() + ctl.simple_command.return_value = {'StateName': 'Pending'} + + with patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') as mock_process: + result = reattach_to_work_unit(job, ctl) + + assert result is False + mock_process.assert_not_called() + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._finalize_adopted_job') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_succeeded_goes_through_process_phase(mock_rmtree, mock_pdd, mock_release, mock_dedup, mock_finalize_job, mock_finalize_run): + """Succeeded units now go through _process_phase so output is not missed. + + The terminal fast-path was removed — all non-Pending/Running units stream + through _process_phase so the dedup can skip already-committed events while + still capturing any output that arrived after the DB snapshot. + """ + job = Mock() + job.id = 42 + job.work_unit_id = 'unit-done' + job.spawned_by_workflow = False + job.started = None + ctl = Mock() + ctl.simple_command.return_value = {'StateName': 'Succeeded', 'ExitCode': 0, 'Detail': ''} + + with patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase', return_value=Mock(status='successful', rc=0)) as mock_process: + reattach_to_work_unit(job, ctl) + + mock_process.assert_called_once() + mock_finalize_job.assert_called_once() + _, _, exit_code, process_phase_failed = mock_finalize_job.call_args[0] + assert exit_code == 0 + assert process_phase_failed is False + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._finalize_adopted_job') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_failed_goes_through_process_phase(mock_rmtree, mock_pdd, mock_release, mock_dedup, mock_finalize_job, mock_finalize_run): + """Failed units go through _process_phase; exit_code=1 from res.status.""" + job = Mock() + job.id = 42 + job.work_unit_id = 'unit-failed' + job.spawned_by_workflow = False + job.started = None + ctl = Mock() + ctl.simple_command.return_value = {'StateName': 'Failed', 'ExitCode': 1, 'Detail': 'exit status 1'} + + with patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase', return_value=Mock(status='failed', rc=1)) as mock_process: + reattach_to_work_unit(job, ctl) + + mock_process.assert_called_once() + mock_finalize_job.assert_called_once() + _, _, exit_code, _ = mock_finalize_job.call_args[0] + assert exit_code == 1 + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._finalize_adopted_job') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_canceled_goes_through_process_phase(mock_rmtree, mock_pdd, mock_release, mock_dedup, mock_finalize_job, mock_finalize_run): + """Canceled units go through _process_phase; exit_code=1 (non-successful).""" + job = Mock() + job.id = 42 + job.work_unit_id = 'unit-canceled' + job.spawned_by_workflow = False + job.started = None + ctl = Mock() + ctl.simple_command.return_value = {'StateName': 'Canceled', 'Detail': ''} + + with patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase', return_value=Mock(status='error', rc=1)) as mock_process: + reattach_to_work_unit(job, ctl) + + mock_process.assert_called_once() + mock_finalize_job.assert_called_once() + _, _, exit_code, _ = mock_finalize_job.call_args[0] + assert exit_code == 1 + + +@patch('awx.main.tasks.jobs._finalize_job_run') +@patch('awx.main.tasks.receptor._compute_adoption_dedup', return_value=(0, set())) +@patch('awx.main.tasks.receptor.AWXReceptorJob._receptor_release_work') +@patch('awx.main.tasks.receptor._get_or_create_private_data_dir', return_value='/tmp/adopt') +@patch('awx.main.tasks.receptor.shutil.rmtree') +def test_reattach_running_defers_without_streaming(mock_rmtree, mock_pdd, mock_release, mock_dedup, mock_finalize): + """Running state returns False — get_work_results blocks on the adopted unit's 0-byte + local stdout file. Defer so adopt_job_async exits and _process_running_jobs retries + next heartbeat; when the EE finishes the unit becomes terminal and is finalized there. + """ + job = Mock() + job.id = 42 + job.work_unit_id = 'unit-running' + job.spawned_by_workflow = False + job.started = None + job.status = 'running' + ctl = Mock() + ctl.simple_command.return_value = {'StateName': 'Running'} + + with patch('awx.main.tasks.receptor.AWXReceptorJob._process_phase') as mock_process: + result = reattach_to_work_unit(job, ctl) + + assert result is False + mock_process.assert_not_called() + + +# --------------------------------------------------------------------------- +# _finalize_adopted_job — all branches +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.jobs._finalize_job_run') +def test_finalize_adopted_job_skips_when_already_finalized(mock_finalize): + """If job.status != 'running' after _process_phase, _finalize_job_run is not called.""" + job = Mock() + job.status = 'successful' + callback = Mock() + + _finalize_adopted_job(job, callback, exit_code=0, process_phase_failed=False) + + mock_finalize.assert_not_called() + + +@patch('awx.main.tasks.jobs._finalize_job_run') +def test_finalize_adopted_job_successful(mock_finalize): + """exit_code=0 → _finalize_job_run called with status='successful' and finished in extra_fields.""" + job = Mock() + job.status = 'running' + job.started = None + callback = Mock() + + _finalize_adopted_job(job, callback, exit_code=0, process_phase_failed=False) + + mock_finalize.assert_called_once() + _, _, _, status, extra_fields = ( + mock_finalize.call_args[0][0], + mock_finalize.call_args[0][1], + mock_finalize.call_args[0][2], + mock_finalize.call_args[0][3], + mock_finalize.call_args[1].get('extra_fields') or mock_finalize.call_args[0][4], + ) + assert status == 'successful' + assert 'finished' in extra_fields + + +@patch('awx.main.tasks.jobs._finalize_job_run') +def test_finalize_adopted_job_failed(mock_finalize): + """exit_code=1 → _finalize_job_run called with status='failed'.""" + job = Mock() + job.status = 'running' + job.started = None + callback = Mock() + + _finalize_adopted_job(job, callback, exit_code=1, process_phase_failed=False) + + _, _, _, status = mock_finalize.call_args[0][:4] + assert status == 'failed' + + +@patch('awx.main.tasks.jobs._finalize_job_run') +def test_finalize_adopted_job_includes_elapsed_when_started(mock_finalize): + """elapsed is passed in extra_fields when job.started is set.""" + from datetime import datetime, timezone + + job = Mock() + job.status = 'running' + job.started = datetime(2026, 1, 1, 0, 0, 0, tzinfo=timezone.utc) + callback = Mock() + + _finalize_adopted_job(job, callback, exit_code=0, process_phase_failed=False) + + extra_fields = mock_finalize.call_args[1].get('extra_fields') or mock_finalize.call_args[0][4] + assert 'elapsed' in extra_fields + + +@patch('awx.main.tasks.jobs._finalize_job_run') +def test_finalize_adopted_job_process_phase_failed_label(mock_finalize): + """process_phase_failed=True still calls _finalize_job_run — just changes the log label.""" + job = Mock() + job.status = 'running' + job.started = None + callback = Mock() + + _finalize_adopted_job(job, callback, exit_code=1, process_phase_failed=True) + + mock_finalize.assert_called_once() + + +# --------------------------------------------------------------------------- +# event_handler dedup_threshold — O(1) integer check +# --------------------------------------------------------------------------- + + +def test_event_handler_dedup_threshold_skips_below_threshold(): + """Events with counter <= dedup_threshold are skipped without hitting persisted_counters.""" + from awx.main.tasks.callback import RunnerCallback + + cb = RunnerCallback(model=None) + cb.dedup_threshold = 10 + cb.persisted_counters = None # not needed — threshold handles it + + dispatched = [] + cb.dispatcher = Mock() + cb.dispatcher.dispatch.side_effect = dispatched.append + + cb.event_handler({'event': 'runner_on_ok', 'counter': 5}) + cb.event_handler({'event': 'runner_on_ok', 'counter': 10}) + assert len(dispatched) == 0 + + +def test_event_handler_dedup_threshold_lets_through_above_threshold(): + """Events above threshold and not in collision zone are let through.""" + from collections import deque + from awx.main.tasks.callback import RunnerCallback + + cb = RunnerCallback(model=None) + cb.dedup_threshold = 5 + cb.persisted_counters = set() + + dispatched = [] + cb.dispatcher = Mock() + cb.dispatcher.dispatch.side_effect = dispatched.append + + # minimal setup to get past the event_handler guards + cb.instance = Mock() + cb.instance.event_class.WRAPUP_EVENT = 'playbook_on_stats' + cb.event_data_key = 'job_id' + cb.job_created = None + cb.parent_workflow_job_id = None + cb.host_map = {} + cb.recent_event_timings = deque(maxlen=100) + + cb.event_handler({'event': 'runner_on_ok', 'counter': 6, 'job_id': 1}) + assert len(dispatched) == 1 + + +def test_event_handler_collision_zone_skips(): + """Events above threshold but in collision_zone are skipped.""" + from awx.main.tasks.callback import RunnerCallback + + cb = RunnerCallback(model=None) + cb.dedup_threshold = 5 + cb.persisted_counters = {7, 8, 9} + + dispatched = [] + cb.dispatcher = Mock() + cb.dispatcher.dispatch.side_effect = dispatched.append + + cb.event_handler({'event': 'runner_on_ok', 'counter': 7}) + cb.event_handler({'event': 'runner_on_ok', 'counter': 9}) + assert len(dispatched) == 0 + + +# --------------------------------------------------------------------------- +# adopt_remote_work — new receptorctl API and JSON fallback +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.receptor.get_tls_client', return_value=None) +@patch('awx.main.tasks.receptor.work_signing_enabled', return_value=False) +def test_adopt_remote_work_uses_adopt_work_when_available(mock_sign, mock_tls): + """If adopt_work() is present on receptor_ctl, call it directly.""" + ctl = Mock() + ctl.adopt_work.return_value = {'unitid': 'u1'} + + result = adopt_remote_work(ctl, node='ee-node', unit_id='u1', config_data={}) + + ctl.adopt_work.assert_called_once_with('ee-node', 'u1', tlsclient=None, signwork=False) + assert result == {'unitid': 'u1'} + + +@patch('awx.main.tasks.receptor.get_tls_client', return_value=None) +@patch('awx.main.tasks.receptor.work_signing_enabled', return_value=False) +def test_adopt_remote_work_json_fallback_without_tls(mock_sign, mock_tls): + """Falls back to raw JSON when adopt_work() is absent and no TLS.""" + ctl = Mock(spec=['connect', 'writestr', 'read_and_parse_json']) # no adopt_work + ctl.read_and_parse_json.return_value = {'unitid': 'u2'} + + adopt_remote_work(ctl, node='ee-node', unit_id='u2', config_data={}) + + ctl.connect.assert_called_once() + payload = json.loads(ctl.writestr.call_args[0][0].rstrip()) + assert payload == {'command': 'work', 'subcommand': 'adopt', 'node': 'ee-node', 'unitid': 'u2'} + assert 'tlsclient' not in payload + assert 'signwork' not in payload + + +@patch('awx.main.tasks.receptor.get_tls_client', return_value='my-tls') +@patch('awx.main.tasks.receptor.work_signing_enabled', return_value=True) +def test_adopt_remote_work_json_fallback_with_tls_and_sign(mock_sign, mock_tls): + """JSON fallback includes tlsclient and signwork when configured.""" + ctl = Mock(spec=['connect', 'writestr', 'read_and_parse_json']) # no adopt_work + ctl.read_and_parse_json.return_value = {} + + adopt_remote_work(ctl, node='ee-node', unit_id='u3', config_data={}) + + payload = json.loads(ctl.writestr.call_args[0][0].rstrip()) + assert payload['tlsclient'] == 'my-tls' + assert payload['signwork'] is True + + +# --------------------------------------------------------------------------- +# reattach_to_work_unit — receptor status check exception returns False +# --------------------------------------------------------------------------- + + +def test_reattach_returns_false_when_status_command_raises(): + """If simple_command('work status ...') raises, return False without crashing.""" + job = Mock() + job.id = 1 + job.work_unit_id = 'dead-unit' + job.created = '2026-01-01T00:00:00Z' + job.spawned_by_workflow = False + + ctl = Mock() + ctl.simple_command.side_effect = Exception('socket closed') + + result = reattach_to_work_unit(job, ctl) + + assert result is False + + +# --------------------------------------------------------------------------- +# _compute_adoption_dedup — collision zone cap warning +# --------------------------------------------------------------------------- + + +@pytest.mark.django_db +def test_compute_adoption_dedup_logs_warning_when_collision_zone_exceeds_cap(caplog): + """When collision_zone_list exceeds the dedup cap, a warning is logged and results are capped.""" + job = Mock(spec=Job) + # Build a queryset-like mock: annotate/filter/values_list chain returns a large list + large_list = list(range(10000)) + + qs = Mock() + qs.annotate.return_value = qs + qs.filter.return_value = qs + qs.order_by.return_value = qs + qs.first.return_value = None # no gap → safe_threshold=0 + qs.values_list.return_value = large_list + + job.get_event_queryset.return_value = qs + + with caplog.at_level('WARNING', logger='awx.main.tasks.receptor'): + safe_threshold, collision_zone = _compute_adoption_dedup(job) + + assert 'collision_zone' in caplog.text + assert safe_threshold == 0 + + +# --------------------------------------------------------------------------- +# _finalize_job_run — extra_fields merges into update fields +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@patch('awx.main.tasks.jobs.ScheduleTaskManager') +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_extra_fields_merged(mock_update, mock_task_mgr, mock_workflow_mgr): + """extra_fields dict is merged into the update kwargs alongside runner fields.""" + instance = Mock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = Mock() + cb.get_delayed_update_fields.return_value = {'emitted_events': 3} + cb.wrapup_event_dispatched = True + + _finalize_job_run(Mock, pk=1, runner_callback=cb, status='successful', extra_fields={'finished': 'NOW', 'elapsed': 1.5}) + + _, kwargs = mock_update.call_args + assert kwargs['finished'] == 'NOW' + assert kwargs['elapsed'] == 1.5 + assert kwargs['emitted_events'] == 3 + + +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_instance_deleted(mock_update): + """When update_model returns None (instance deleted), returns None early.""" + cb = Mock() + cb.get_delayed_update_fields.return_value = {} + mock_update.return_value = None + + result = _finalize_job_run(Mock, pk=1, runner_callback=cb, status='successful') + + assert result is None + + +@patch('awx.main.tasks.jobs.update_inventory_computed_fields') +@patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@patch('awx.main.tasks.jobs.ScheduleTaskManager') +@patch('awx.main.tasks.jobs.events_processed_hook') +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_with_blocked_jobs(mock_update, mock_hook, mock_task_mgr, mock_wf_mgr, mock_inv_update): + """ScheduleTaskManager is called when unifiedjob_blocked_jobs.exists().""" + instance = Mock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = True + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = Mock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + _finalize_job_run(Mock, pk=1, runner_callback=cb, status='successful') + + mock_task_mgr.return_value.schedule.assert_called_once() + mock_wf_mgr.return_value.schedule.assert_not_called() + + +@patch('awx.main.tasks.jobs.update_inventory_computed_fields') +@patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@patch('awx.main.tasks.jobs.ScheduleTaskManager') +@patch('awx.main.tasks.jobs.events_processed_hook') +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_with_workflow(mock_update, mock_hook, mock_task_mgr, mock_wf_mgr, mock_inv_update): + """ScheduleWorkflowManager is called when spawned_by_workflow is True.""" + instance = Mock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = True + instance.inventory_id = None + mock_update.return_value = instance + + cb = Mock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + _finalize_job_run(Mock, pk=1, runner_callback=cb, status='successful') + + mock_wf_mgr.return_value.schedule.assert_called_once() + mock_task_mgr.return_value.schedule.assert_not_called() + + +@patch('awx.main.tasks.jobs.update_inventory_computed_fields') +@patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@patch('awx.main.tasks.jobs.ScheduleTaskManager') +@patch('awx.main.tasks.jobs.events_processed_hook') +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_with_inventory(mock_update, mock_hook, mock_task_mgr, mock_wf_mgr, mock_inv_update): + """update_inventory_computed_fields.delay is called when inventory_id is set.""" + instance = Mock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = 42 + mock_update.return_value = instance + + cb = Mock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + _finalize_job_run(Mock, pk=1, runner_callback=cb, status='successful') + + mock_inv_update.delay.assert_called_once_with(42) + + +@patch('awx.main.tasks.jobs.update_inventory_computed_fields') +@patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@patch('awx.main.tasks.jobs.ScheduleTaskManager') +@patch('awx.main.tasks.jobs.events_processed_hook') +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_inventory_delay_exception(mock_update, mock_hook, mock_task_mgr, mock_wf_mgr, mock_inv_update, caplog): + """Exception in inventory delay is logged and finalization continues.""" + instance = Mock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = 42 + instance.log_format = 'job 1' + mock_update.return_value = instance + mock_inv_update.delay.side_effect = RuntimeError('celery error') + + cb = Mock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + result = _finalize_job_run(Mock, pk=1, runner_callback=cb, status='successful') + + assert result is instance + assert 'Error scheduling inventory computed fields update' in caplog.text + + +@patch('awx.main.tasks.jobs.update_inventory_computed_fields') +@patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@patch('awx.main.tasks.jobs.ScheduleTaskManager') +@patch('awx.main.tasks.jobs.events_processed_hook') +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_calls_hook_on_host_counts(mock_update, mock_hook, mock_task_mgr, mock_wf_mgr, mock_inv_update): + """events_processed_hook is called when host_status_counts is not None.""" + instance = Mock() + instance.host_status_counts = {'ok': 1, 'failed': 0} + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = Mock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + _finalize_job_run(Mock, pk=1, runner_callback=cb, status='successful') + + mock_hook.assert_called_once_with(instance) + + +@patch('awx.main.tasks.jobs.update_inventory_computed_fields') +@patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@patch('awx.main.tasks.jobs.ScheduleTaskManager') +@patch('awx.main.tasks.jobs.events_processed_hook') +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_calls_hook_on_wrapup_not_dispatched(mock_update, mock_hook, mock_task_mgr, mock_wf_mgr, mock_inv_update): + """events_processed_hook is called when wrapup_event_dispatched is False.""" + instance = Mock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = Mock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = False + + _finalize_job_run(Mock, pk=1, runner_callback=cb, status='successful') + + mock_hook.assert_called_once_with(instance) + + +@patch('awx.main.tasks.jobs.update_inventory_computed_fields') +@patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@patch('awx.main.tasks.jobs.ScheduleTaskManager') +@patch('awx.main.tasks.jobs.events_processed_hook') +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_skips_hook_when_no_counts_and_wrapup(mock_update, mock_hook, mock_task_mgr, mock_wf_mgr, mock_inv_update): + """events_processed_hook is NOT called when counts=None AND wrapup_dispatched=True.""" + instance = Mock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = Mock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + _finalize_job_run(Mock, pk=1, runner_callback=cb, status='successful') + + mock_hook.assert_not_called() + + +@patch('awx.main.tasks.jobs.update_inventory_computed_fields') +@patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@patch('awx.main.tasks.jobs.ScheduleTaskManager') +@patch('awx.main.tasks.jobs.events_processed_hook') +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_calls_websocket_emit(mock_update, mock_hook, mock_task_mgr, mock_wf_mgr, mock_inv_update): + """websocket_emit_status is called with the status.""" + instance = Mock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = Mock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + _finalize_job_run(Mock, pk=1, runner_callback=cb, status='failed') + + instance.websocket_emit_status.assert_called_once_with('failed') + + +@patch('awx.main.tasks.jobs.update_inventory_computed_fields') +@patch('awx.main.tasks.jobs.ScheduleWorkflowManager') +@patch('awx.main.tasks.jobs.ScheduleTaskManager') +@patch('awx.main.tasks.jobs.events_processed_hook') +@patch('awx.main.tasks.jobs.update_model') +def test_finalize_job_run_logs_lifecycle(mock_update, mock_hook, mock_task_mgr, mock_wf_mgr, mock_inv_update): + """log_lifecycle('finalize_run') is called.""" + instance = Mock() + instance.host_status_counts = None + instance.unifiedjob_blocked_jobs.exists.return_value = False + instance.spawned_by_workflow = False + instance.inventory_id = None + mock_update.return_value = instance + + cb = Mock() + cb.get_delayed_update_fields.return_value = {} + cb.wrapup_event_dispatched = True + + _finalize_job_run(Mock, pk=1, runner_callback=cb, status='successful') + + instance.log_lifecycle.assert_called_once_with('finalize_run') + + +# --------------------------------------------------------------------------- +# _configure_runner_callback — host_map exception swallowed +# --------------------------------------------------------------------------- + + +def test_configure_runner_callback_populates_host_map(): + """configure_for_job populates host_map from inventory hostvars.""" + cb = RunnerCallback(model=None) + instance = Mock() + instance.created = '2026-01-01' + instance.spawned_by_workflow = False + instance.inventory_id = 42 + instance.inventory.get_script_data.return_value = { + '_meta': { + 'hostvars': { + 'host1': {'remote_tower_id': 'ext-1'}, + 'host2': {'remote_tower_id': 'ext-2'}, + } + } + } + + cb.configure_for_job(instance) + + assert cb.host_map == {'host1': 'ext-1', 'host2': 'ext-2'} + + +def test_configure_runner_callback_swallows_host_map_error(): + """If fetching inventory hosts raises, the exception is swallowed and host_map stays {}.""" + cb = RunnerCallback(model=None) + instance = Mock() + instance.created = '2026-01-01' + instance.spawned_by_workflow = False + instance.inventory_id = 1 + instance.inventory.get_script_data.side_effect = Exception('DB error') + + cb.configure_for_job(instance) # must not raise + + assert cb.host_map == {} + + +# --------------------------------------------------------------------------- +# _handle_work_error — edge cases and error paths +# --------------------------------------------------------------------------- + + +@patch('awx.main.tasks.receptor.logger') +def test_handle_work_error_no_output_no_detail(mock_logger): + """When there's no receptor output and no detail, warning is logged.""" + res = namedtuple('result', ['status', 'rc'])('error', 1) + receptor_ctl = Mock() + receptor_ctl.simple_command.return_value = { + 'StateName': 'Succeeded', # Not 'Failed', so no attempt to fetch work results + 'Detail': '', # Empty detail + 'StdoutSize': 0, + } + + receptor_job = AWXReceptorJob.__new__(AWXReceptorJob) + receptor_job.unit_id = 'test-unit' + receptor_job.task = Mock() + receptor_job.task.instance = Mock() + receptor_job.task.instance.log_format = 'job 1' + receptor_job.task.runner_callback = Mock() + receptor_job.task.runner_callback.extra_update_fields = {} + receptor_job.task.runner_callback.event_ct = 0 # No events + + result = receptor_job._handle_work_error(receptor_ctl, res) + + assert result == res + mock_logger.warning.assert_called() + # Verify warning was logged for "No result details or output" + assert any('No result details' in str(call) for call in mock_logger.warning.call_args_list) + + +@patch('awx.main.tasks.receptor.logger') +def test_handle_work_error_state_name_empty_string(mock_logger): + """When state_name can't be determined, error path handles empty string gracefully.""" + res = namedtuple('result', ['status', 'rc'])('error', 1) + receptor_ctl = Mock() + # Simulate status command failure + receptor_ctl.simple_command.side_effect = Exception('connection lost') + + receptor_job = AWXReceptorJob.__new__(AWXReceptorJob) + receptor_job.unit_id = 'test-unit' + receptor_job.task = Mock() + receptor_job.task.instance = Mock() + receptor_job.task.instance.log_format = 'job 1' + receptor_job.task.runner_callback = Mock() + receptor_job.task.runner_callback.extra_update_fields = {} + receptor_job.task.runner_callback.event_ct = 5 + + result = receptor_job._handle_work_error(receptor_ctl, res) + + assert result == res + # Verify exception was logged + mock_logger.exception.assert_called() + + +def test_handle_work_error_delay_update_with_receptor_output(): + """When receptor_output is collected, it's passed to delay_update.""" + res = namedtuple('result', ['status', 'rc'])('error', 1) + receptor_ctl = Mock() + resultfile = Mock() + resultfile.readlines.return_value = [b'Error: connection refused\n'] + resultfile.close = Mock() + resultsock = Mock() + + receptor_ctl.simple_command.return_value = { + 'StateName': 'Failed', + 'Detail': '', + 'StdoutSize': 500, + } + receptor_ctl.get_work_results.return_value = (resultsock, resultfile) + + receptor_job = AWXReceptorJob.__new__(AWXReceptorJob) + receptor_job.unit_id = 'test-unit' + receptor_job.task = Mock() + receptor_job.task.instance = Mock() + receptor_job.task.instance.log_format = 'job 1' + receptor_job.task.runner_callback = Mock() + receptor_job.task.runner_callback.extra_update_fields = {} + receptor_job.task.runner_callback.event_ct = 0 # No events → fetch output + + result = receptor_job._handle_work_error(receptor_ctl, res) + + assert result == res + # Verify delay_update was called with receptor output + receptor_job.task.runner_callback.delay_update.assert_called_once() + call_kwargs = receptor_job.task.runner_callback.delay_update.call_args[1] + assert 'Worker output' in call_kwargs['result_traceback'] + assert 'connection refused' in call_kwargs['result_traceback'] + + +def test_handle_work_error_delay_update_with_detail_fallback(): + """When there's no receptor output but detail is present, detail is used.""" + res = namedtuple('result', ['status', 'rc'])('error', 1) + receptor_ctl = Mock() + receptor_ctl.simple_command.return_value = { + 'StateName': 'Succeeded', # Not 'Failed', so no attempt to fetch output + 'Detail': 'Permission denied', + 'StdoutSize': 100, + } + + receptor_job = AWXReceptorJob.__new__(AWXReceptorJob) + receptor_job.unit_id = 'test-unit' + receptor_job.task = Mock() + receptor_job.task.instance = Mock() + receptor_job.task.instance.log_format = 'job 1' + receptor_job.task.runner_callback = Mock() + receptor_job.task.runner_callback.extra_update_fields = {} + receptor_job.task.runner_callback.event_ct = 5 + + result = receptor_job._handle_work_error(receptor_ctl, res) + + assert result == res + # Verify delay_update was called with detail + receptor_job.task.runner_callback.delay_update.assert_called_once() + call_kwargs = receptor_job.task.runner_callback.delay_update.call_args[1] + assert 'Receptor detail' in call_kwargs['result_traceback'] + assert 'Permission denied' in call_kwargs['result_traceback'] diff --git a/awx/main/tests/unit/test_tasks.py b/awx/main/tests/unit/test_tasks.py index cdc98bacac..278f7c3437 100644 --- a/awx/main/tests/unit/test_tasks.py +++ b/awx/main/tests/unit/test_tasks.py @@ -469,11 +469,12 @@ class TestGenericRun: task.model.objects.get = mock.Mock(return_value=job) task.build_private_data_files = mock.Mock(side_effect=OSError()) - with mock.patch('awx.main.tasks.jobs.shutil.copytree'), mock.patch('awx.main.tasks.jobs.evaluate_policy'): - with pytest.raises(Exception): - task.run(1) + with mock.patch('awx.main.tasks.jobs.update_model', return_value=job) as mock_update_model: + with mock.patch('awx.main.tasks.jobs.shutil.copytree'), mock.patch('awx.main.tasks.jobs.evaluate_policy'): + with pytest.raises(Exception): + task.run(1) - update_model_call = task.update_model.call_args[1] + update_model_call = mock_update_model.call_args[1] assert 'OSError' in update_model_call['result_traceback'] assert update_model_call['status'] == 'error' assert update_model_call['emitted_events'] == 0 @@ -574,11 +575,12 @@ class TestAdhocRun(TestJobExecution): task.model.objects.get = mock.Mock(return_value=adhoc_job) task.build_inventory = mock.Mock() - with pytest.raises(Exception): - task.run(adhoc_job.pk) + with mock.patch('awx.main.tasks.jobs.update_model', return_value=adhoc_job) as mock_update_model: + with pytest.raises(Exception): + task.run(adhoc_job.pk) call_args, _ = task.update_model.call_args_list[0] - update_model_call = task.update_model.call_args[1] + update_model_call = mock_update_model.call_args[1] assert 'Jinja variables are not allowed' in update_model_call['result_traceback'] ''' diff --git a/awx/settings/defaults.py b/awx/settings/defaults.py index 8143e306df..43d7bac0ca 100644 --- a/awx/settings/defaults.py +++ b/awx/settings/defaults.py @@ -223,6 +223,9 @@ JOB_EVENT_WORKERS = 4 # Minimum number of workers for the dispatcher (dispatcherd) process pool DISPATCHER_MIN_WORKERS = 4 +# Maximum number of events buffered per callback worker before flushing via bulk_create() +JOB_EVENT_CALLBACK_BUFFER_SIZE = 1000 + # The number of seconds to buffer callback receiver bulk # writes in memory before flushing via JobEvent.objects.bulk_create() JOB_EVENT_BUFFER_SECONDS = 1 @@ -418,6 +421,12 @@ CLUSTER_NODE_HEARTBEAT_PERIOD = 60 CLUSTER_NODE_MISSED_HEARTBEAT_TOLERANCE = 2 RECEPTOR_SERVICE_ADVERTISEMENT_PERIOD = 60 # https://github.com/ansible/receptor/blob/aa1d589e154d8a0cb99a220aff8f98faf2273be6/pkg/netceptor/netceptor.go#L34 + +# Seconds after a controller restart before an orphaned dispatched job is failed. +# Jobs with work_unit_id set are skipped by the reaper and handled by the adoption loop +# (_attempt_adoption_for_dispatched_jobs). This timeout is the maximum time we wait for +# the EE to finish before giving up and failing the job. +HADR_JOB_ADOPTION_TIMEOUT = 3600 EXECUTION_NODE_REMEDIATION_CHECKS = 60 * 30 # once every 30 minutes check if an execution node errors have been resolved # Amount of time dispatcher will try to reconnect to database for jobs and consuming new work