diff --git a/awx/main/tasks/adoption_decisions.py b/awx/main/tasks/adoption_decisions.py new file mode 100644 index 0000000000..d7da81d679 --- /dev/null +++ b/awx/main/tasks/adoption_decisions.py @@ -0,0 +1,110 @@ +"""Decisions the heartbeat makes about lost instances and orphaned jobs. + +Nothing in this module touches the database, receptor or the dispatcher. The heartbeat in +awx.main.tasks.system gathers state, asks these functions what to do, and carries out the +answer. Keeping the decisions separate lets them be tested with plain objects, without a +cluster, and gives cross-controller adoption (AAP-89602) one place to change each rule. + +Jobs and instances are read only through attributes, so model instances and simple +stand-ins in tests both work. +""" + +from enum import Enum + +from django.utils.timezone import timedelta + + +class JobAction(str, Enum): + SKIP = 'skip' + ADOPT = 'adopt' + REAP = 'reap' + + +class LostInstanceAction(str, Enum): + DEPROVISION = 'deprovision' + MARK_OFFLINE = 'mark_offline' + NONE = 'none' + + +def find_lost_instances(instance_list, this_hostname, ref_time): + """Split peers into (lost, remaining). This instance is never lost here; it is judged separately.""" + lost = [] + remaining = [] + for inst in instance_list: + if inst.hostname != this_hostname and inst.is_lost(ref_time=ref_time): + lost.append(inst) + else: + remaining.append(inst) + return lost, remaining + + +def gate_lost_instances(lost_instances, mesh_ready): + """Return the lost instances that may be handled now. + + While the receptor mesh is re-establishing, a control node may only look lost, so its + handling waits for a later heartbeat. Execution and hop nodes don't depend on mesh + consensus and are handled either way. + """ + if mesh_ready: + return list(lost_instances) + return [inst for inst in lost_instances if inst.node_type in ('execution', 'hop')] + + +def lost_instance_job_action(job): + """What to do with a running job whose controller or execution node was lost. + + AAP-89602: cross-controller adoption for dispatched jobs will return ADOPT here + once ansible/receptor#1564 merges. Until then every such job is reaped. + """ + return JobAction.REAP + + +def lost_instance_disposition(inst, auto_deprovision): + """What to do with the lost instance itself once its jobs are handled.""" + if auto_deprovision and inst.node_type == 'control': + return LostInstanceAction.DEPROVISION + if inst.node_state == 'ready': + return LostInstanceAction.MARK_OFFLINE + return LostInstanceAction.NONE + + +def startup_job_action(job): + """At startup, a running job this controller owns is adopted if it was dispatched to receptor, else reaped.""" + if job.work_unit_id: + return JobAction.ADOPT + return JobAction.REAP + + +def running_job_action(job, this_hostname, active_task_ids, ref_time, workflow_ctype_id): + """On a periodic heartbeat, what to do with a job that may have been orphaned on this instance. + + A job is left alone unless it is running, belongs to this instance as controller or + execution node, is not a workflow, started before ref_time, and no dispatcher task is + tracking it. Such an orphan is adopted when this instance controls it and it was + dispatched to receptor; otherwise it is reaped. + """ + if job.status != 'running': + return JobAction.SKIP + if this_hostname not in (job.controller_node, job.execution_node): + return JobAction.SKIP + if job.polymorphic_ctype_id == workflow_ctype_id: + return JobAction.SKIP + if active_task_ids and job.celery_task_id in active_task_ids: + return JobAction.SKIP + if ref_time and (job.started is None or job.started > ref_time): + return JobAction.SKIP + if job.work_unit_id and job.controller_node == this_hostname: + return JobAction.ADOPT + return JobAction.REAP + + +def adoption_deadline_passed(last_event_time, started, ref_time, timeout_seconds): + """True if a job has been silent too long to adopt. + + Silence is measured from the last saved event, or from the start when there are no + events. A job with neither is never past the deadline. + """ + orphaned_since = last_event_time or started + if not orphaned_since: + return False + return orphaned_since < ref_time - timedelta(seconds=timeout_seconds) diff --git a/awx/main/tasks/system.py b/awx/main/tasks/system.py index 47b3823a1e..51e27dd61d 100644 --- a/awx/main/tasks/system.py +++ b/awx/main/tasks/system.py @@ -71,6 +71,17 @@ from awx.main.models import ( convert_jsonfields, ) from awx.main.models.credential import CredentialType +from awx.main.tasks.adoption_decisions import ( + JobAction, + LostInstanceAction, + adoption_deadline_passed, + find_lost_instances, + gate_lost_instances, + lost_instance_disposition, + lost_instance_job_action, + running_job_action, + startup_job_action, +) from awx.main.tasks.helpers import is_run_threshold_reached from awx.main.tasks.host_indirect import save_indirect_host_entries from awx.main.tasks.receptor import ( @@ -771,7 +782,6 @@ def _heartbeat_instance_management(): nowtime = now() instance_list = list(Instance.objects.filter(node_state__in=(Instance.States.READY, Instance.States.UNAVAILABLE, Instance.States.INSTALLED))) this_inst = None - lost_instances = [] for inst in instance_list: if inst.hostname == settings.CLUSTER_HOST_ID: @@ -795,12 +805,7 @@ def _heartbeat_instance_management(): inspect_execution_and_hop_nodes(instance_list, mesh_status) - for inst in list(instance_list): - if inst == this_inst: - continue - if inst.is_lost(ref_time=nowtime): - lost_instances.append(inst) - instance_list.remove(inst) + lost_instances, instance_list = find_lost_instances(instance_list, settings.CLUSTER_HOST_ID, nowtime) if this_inst: startup_event = this_inst.is_lost(ref_time=nowtime) @@ -823,11 +828,8 @@ def _heartbeat_instance_management(): logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID)) 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, ctl + if lost_instances: + lost_instances = gate_lost_instances(lost_instances, _mesh_all_ready_nodes_visible(mesh_status)) return this_inst, instance_list, lost_instances, ctl @@ -862,19 +864,19 @@ def _reap_and_mark_lost_instance(other_inst): ).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') + if lost_instance_job_action(j) == JobAction.REAP: + 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: logger.exception('failed to re-process jobs for lost instance {}'.format(other_inst.hostname)) try: - if settings.AWX_AUTO_DEPROVISION_INSTANCES and other_inst.node_type == "control": + disposition = lost_instance_disposition(other_inst, settings.AWX_AUTO_DEPROVISION_INSTANCES) + if disposition == LostInstanceAction.DEPROVISION: deprovision_hostname = other_inst.hostname other_inst.delete() # FIXME: what about associated inbound links? logger.info("Host {} Automatically Deprovisioned.".format(deprovision_hostname)) - elif other_inst.node_state == Instance.States.READY: + elif disposition == LostInstanceAction.MARK_OFFLINE: other_inst.mark_offline(errors=_('Another cluster node has determined this instance to be unresponsive')) logger.error("Host {} last checked in at {}, marked as lost.".format(other_inst.hostname, other_inst.last_seen)) @@ -950,7 +952,7 @@ def _process_startup_jobs(this_inst): reaped_ids = [] for j in jobs: try: - if j.work_unit_id: + if startup_job_action(j) == JobAction.ADOPT: 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: @@ -975,6 +977,7 @@ def _process_running_jobs(this_inst, active_task_ids, ref_time): - Undispatched or not owned by this controller → reap. Cross-controller adoption (jobs from a dead peer controller) is deferred to AAP-89602. + The query narrows the candidates; running_job_action makes the decision for each one. """ 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)) @@ -989,12 +992,13 @@ def _process_running_jobs(this_inst, active_task_ids, ref_time): for j in jobs: try: - if j.work_unit_id and j.controller_node == this_inst.hostname: + action = running_job_action(j, this_inst.hostname, active_task_ids, ref_time, workflow_ctype_id) + if action == JobAction.ADOPT: 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: + elif action == JobAction.REAP: reaper.reap_job(j, 'failed') except Exception: logger.exception(f'Failed processing job {j.id} in heartbeat job loop') @@ -1014,10 +1018,8 @@ def adopt_job_async(job_id): 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: + if adoption_deadline_passed(last_event_time, job.started, now(), adoption_timeout): 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 diff --git a/awx/main/tests/unit/tasks/test_adoption_decisions.py b/awx/main/tests/unit/tasks/test_adoption_decisions.py new file mode 100644 index 0000000000..a15eb2fca1 --- /dev/null +++ b/awx/main/tests/unit/tasks/test_adoption_decisions.py @@ -0,0 +1,207 @@ +from datetime import datetime, timezone +from types import SimpleNamespace + +import pytest +from django.utils.timezone import timedelta + +from awx.main.models import Instance +from awx.main.tasks.adoption_decisions import ( + JobAction, + LostInstanceAction, + adoption_deadline_passed, + find_lost_instances, + gate_lost_instances, + lost_instance_disposition, + lost_instance_job_action, + running_job_action, + startup_job_action, +) + +NOW = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc) +ME = 'awx-1' +PEER = 'awx-2' +EXEC = 'receptor-1' +WORKFLOW_CTYPE = 99 +JOB_CTYPE = 1 + + +def make_job(**kwargs): + fields = dict( + status='running', + controller_node=ME, + execution_node=EXEC, + polymorphic_ctype_id=JOB_CTYPE, + celery_task_id='task-1', + started=NOW - timedelta(minutes=5), + work_unit_id='unit-1', + ) + fields.update(kwargs) + return SimpleNamespace(**fields) + + +def make_instance(hostname, node_type='control', seconds_since_seen=0, node_state='ready'): + last_seen = None if seconds_since_seen is None else NOW - timedelta(seconds=seconds_since_seen) + return Instance(hostname=hostname, node_type=node_type, node_state=node_state, last_seen=last_seen) + + +@pytest.fixture +def heartbeat_settings(settings): + settings.CLUSTER_NODE_HEARTBEAT_PERIOD = 60 + settings.CLUSTER_NODE_MISSED_HEARTBEAT_TOLERANCE = 2 + settings.RECEPTOR_SERVICE_ADVERTISEMENT_PERIOD = 60 + return settings + + +class TestFindLostInstances: + def test_peer_within_grace_period_is_not_lost(self, heartbeat_settings): + peer = make_instance(PEER, seconds_since_seen=119) + lost, remaining = find_lost_instances([peer], ME, NOW) + assert lost == [] + assert remaining == [peer] + + def test_peer_past_grace_period_is_lost(self, heartbeat_settings): + peer = make_instance(PEER, seconds_since_seen=121) + lost, remaining = find_lost_instances([peer], ME, NOW) + assert lost == [peer] + assert remaining == [] + + def test_peer_never_seen_is_lost(self, heartbeat_settings): + peer = make_instance(PEER, seconds_since_seen=None) + lost, _ = find_lost_instances([peer], ME, NOW) + assert lost == [peer] + + def test_this_instance_is_never_lost_here(self, heartbeat_settings): + me = make_instance(ME, seconds_since_seen=10_000) + lost, remaining = find_lost_instances([me], ME, NOW) + assert lost == [] + assert remaining == [me] + + def test_execution_node_gets_advertisement_period_added(self, heartbeat_settings): + node = make_instance(EXEC, node_type='execution', seconds_since_seen=150) + lost, _ = find_lost_instances([node], ME, NOW) + assert lost == [] + + def test_slow_controller_is_judged_lost_while_still_running(self, heartbeat_settings): + # A controller whose heartbeat is delayed past the grace period looks exactly like + # a dead one. The decision has only last_seen to go on. + slow = make_instance(PEER, seconds_since_seen=125) + lost, _ = find_lost_instances([slow], ME, NOW) + assert lost == [slow] + + def test_clock_skew_on_the_judge_makes_a_healthy_peer_look_lost(self, heartbeat_settings): + # The peer heartbeated just now by its own clock, but this instance's clock runs + # five minutes ahead. + peer = make_instance(PEER, seconds_since_seen=0) + lost, _ = find_lost_instances([peer], ME, NOW + timedelta(minutes=5)) + assert lost == [peer] + + +class TestGateLostInstances: + def test_mesh_ready_handles_every_lost_instance(self): + lost = [make_instance(PEER), make_instance(EXEC, node_type='execution'), make_instance('hop-1', node_type='hop')] + assert gate_lost_instances(lost, mesh_ready=True) == lost + + def test_mesh_not_ready_defers_only_control_nodes(self): + control = make_instance(PEER) + execution = make_instance(EXEC, node_type='execution') + hop = make_instance('hop-1', node_type='hop') + hybrid = make_instance('hybrid-1', node_type='hybrid') + assert gate_lost_instances([control, execution, hop, hybrid], mesh_ready=False) == [execution, hop] + + +class TestLostInstanceDecisions: + @pytest.mark.parametrize('work_unit_id', ['unit-1', '']) + def test_jobs_of_a_lost_instance_are_reaped(self, work_unit_id): + assert lost_instance_job_action(make_job(controller_node=PEER, work_unit_id=work_unit_id)) == JobAction.REAP + + @pytest.mark.parametrize( + 'node_type, node_state, auto_deprovision, expected', + [ + ('control', 'ready', True, LostInstanceAction.DEPROVISION), + ('control', 'unavailable', True, LostInstanceAction.DEPROVISION), + ('execution', 'ready', True, LostInstanceAction.MARK_OFFLINE), + ('control', 'ready', False, LostInstanceAction.MARK_OFFLINE), + ('control', 'unavailable', False, LostInstanceAction.NONE), + ('control', 'installed', False, LostInstanceAction.NONE), + ('execution', 'unavailable', True, LostInstanceAction.NONE), + ], + ) + def test_disposition(self, node_type, node_state, auto_deprovision, expected): + inst = make_instance(PEER, node_type=node_type, node_state=node_state) + assert lost_instance_disposition(inst, auto_deprovision) == expected + + +class TestStartupJobAction: + def test_dispatched_job_is_adopted(self): + assert startup_job_action(make_job(work_unit_id='unit-1')) == JobAction.ADOPT + + @pytest.mark.parametrize('work_unit_id', ['', None]) + def test_undispatched_job_is_reaped(self, work_unit_id): + assert startup_job_action(make_job(work_unit_id=work_unit_id)) == JobAction.REAP + + +class TestRunningJobAction: + def decide(self, job, active_task_ids=('other-task',), ref_time=NOW): + return running_job_action(job, ME, list(active_task_ids), ref_time, WORKFLOW_CTYPE) + + def test_orphan_this_instance_controls_is_adopted(self): + assert self.decide(make_job()) == JobAction.ADOPT + + def test_job_a_dispatcher_task_still_tracks_is_left_alone(self): + assert self.decide(make_job(celery_task_id='task-1'), active_task_ids=['task-1']) == JobAction.SKIP + + def test_orphan_not_yet_dispatched_to_receptor_is_reaped(self): + assert self.decide(make_job(work_unit_id='')) == JobAction.REAP + + def test_orphan_this_instance_only_executes_is_reaped(self): + # A hybrid node running a job for another controller: it can't adopt, so it reaps. + assert self.decide(make_job(controller_node=PEER, execution_node=ME)) == JobAction.REAP + + def test_job_of_another_instance_is_left_alone(self): + assert self.decide(make_job(controller_node=PEER)) == JobAction.SKIP + + @pytest.mark.parametrize('status', ['pending', 'waiting', 'successful', 'failed', 'canceled']) + def test_job_not_running_is_left_alone(self, status): + assert self.decide(make_job(status=status)) == JobAction.SKIP + + def test_workflow_job_is_left_alone(self): + assert self.decide(make_job(polymorphic_ctype_id=WORKFLOW_CTYPE)) == JobAction.SKIP + + def test_job_started_after_ref_time_is_left_alone(self): + assert self.decide(make_job(started=NOW + timedelta(seconds=1))) == JobAction.SKIP + + def test_job_without_start_time_is_left_alone(self): + assert self.decide(make_job(started=None)) == JobAction.SKIP + + def test_no_ref_time_considers_every_start_time(self): + assert self.decide(make_job(started=NOW + timedelta(days=1)), ref_time=None) == JobAction.ADOPT + + def test_empty_active_list_treats_every_job_as_orphaned(self): + assert self.decide(make_job(), active_task_ids=()) == JobAction.ADOPT + + +class TestAdoptionDeadline: + TIMEOUT = 300 + + def test_recent_event_is_within_deadline(self): + assert not adoption_deadline_passed(NOW - timedelta(seconds=299), NOW - timedelta(hours=1), NOW, self.TIMEOUT) + + def test_old_last_event_is_past_deadline(self): + assert adoption_deadline_passed(NOW - timedelta(seconds=301), NOW - timedelta(hours=1), NOW, self.TIMEOUT) + + def test_last_event_wins_over_start_time(self): + # A long job that is still producing events is not timed out by its age. + assert not adoption_deadline_passed(NOW - timedelta(seconds=10), NOW - timedelta(days=1), NOW, self.TIMEOUT) + + def test_start_time_used_when_no_events(self): + assert adoption_deadline_passed(None, NOW - timedelta(seconds=301), NOW, self.TIMEOUT) + assert not adoption_deadline_passed(None, NOW - timedelta(seconds=299), NOW, self.TIMEOUT) + + def test_job_with_no_events_or_start_is_never_past_deadline(self): + assert not adoption_deadline_passed(None, None, NOW, self.TIMEOUT) + + def test_quiet_job_is_past_deadline_even_while_healthy(self): + # A playbook task that runs silently longer than the timeout (a long pause, a slow + # package install) looks the same as a dead one. + last_event = NOW - timedelta(seconds=self.TIMEOUT + 1) + assert adoption_deadline_passed(last_event, NOW - timedelta(seconds=self.TIMEOUT + 60), NOW, self.TIMEOUT)