diff --git a/awx/main/tasks/system.py b/awx/main/tasks/system.py index 9d39dda96c..9554a8d05a 100644 --- a/awx/main/tasks/system.py +++ b/awx/main/tasks/system.py @@ -788,17 +788,48 @@ def _mesh_all_ready_nodes_visible(mesh_status): return True +def _receptor_mesh_status(ctl): + """Return receptor's status, or None when it can't be read.""" + try: + return ctl.simple_command('status') + except (OSError, RuntimeError, ValueError) as exc: + logger.warning(f'Receptor status unavailable: {exc}') + return None + + +def _check_in_this_instance(this_inst, nowtime): + """Run this instance's health check. Return False when it is rejoining the cluster.""" + startup_event = this_inst.is_lost(ref_time=nowtime) + last_last_seen = this_inst.last_seen + 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 False + if 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): + logger.warning(f'Heartbeat skew - interval={(nowtime - last_last_seen).total_seconds():.4f}, expected={settings.CLUSTER_NODE_HEARTBEAT_PERIOD}') + return True + + +def _register_missing_this_instance(): + """Recreate this instance's record when auto-deprovisioning removed it. Return None when it can't.""" + if not settings.AWX_AUTO_DEPROVISION_INSTANCES: + logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID)) + return None + changed, this_inst = Instance.objects.register(ip_address=os.environ.get('MY_POD_IP'), node_type='control', node_uuid=settings.SYSTEM_UUID) + if changed: + logger.warning(f'Recreated instance record {this_inst.hostname} after unexpected removal') + this_inst.local_health_check() + return this_inst + + def _heartbeat_instance_management(clock=None): """Common logic for heartbeat instance management.""" logger.debug("Cluster node heartbeat task.") nowtime = (clock or now)() instance_list = list(Instance.objects.filter(node_state__in=(Instance.States.READY, Instance.States.UNAVAILABLE, Instance.States.INSTALLED))) - this_inst = None - - for inst in instance_list: - if inst.hostname == settings.CLUSTER_HOST_ID: - this_inst = inst - break + this_inst = next((inst for inst in instance_list if inst.hostname == settings.CLUSTER_HOST_ID), None) try: ctl = get_receptor_ctl() @@ -809,35 +840,18 @@ def _heartbeat_instance_management(clock=None): this_inst.mark_offline(errors='Receptor not available') return None, None, None, None - try: - mesh_status = ctl.simple_command('status') - except (OSError, RuntimeError, ValueError) as exc: - logger.warning(f'Receptor status unavailable: {exc}') - mesh_status = None + mesh_status = _receptor_mesh_status(ctl) inspect_execution_and_hop_nodes(instance_list, mesh_status) 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) - last_last_seen = this_inst.last_seen - 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}') + if not _check_in_this_instance(this_inst, nowtime): 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): - logger.warning(f'Heartbeat skew - interval={(nowtime - last_last_seen).total_seconds():.4f}, expected={settings.CLUSTER_NODE_HEARTBEAT_PERIOD}') else: - if settings.AWX_AUTO_DEPROVISION_INSTANCES: - changed, this_inst = Instance.objects.register(ip_address=os.environ.get('MY_POD_IP'), node_type='control', node_uuid=settings.SYSTEM_UUID) - if changed: - logger.warning(f'Recreated instance record {this_inst.hostname} after unexpected removal') - this_inst.local_health_check() - else: - logger.error("Cluster Host Not Found: {}".format(settings.CLUSTER_HOST_ID)) + this_inst = _register_missing_this_instance() + if this_inst is None: return None, None, None, None lost_instances = gate_lost_instances(lost_instances, lambda: _mesh_all_ready_nodes_visible(mesh_status)) diff --git a/awx/main/tests/functional/tasks/test_tasks_system.py b/awx/main/tests/functional/tasks/test_tasks_system.py index 3b589e5a7c..add20f4fb2 100644 --- a/awx/main/tests/functional/tasks/test_tasks_system.py +++ b/awx/main/tests/functional/tasks/test_tasks_system.py @@ -676,6 +676,18 @@ def test_heartbeat_marks_offline_when_receptor_unavailable(settings): assert this_inst.node_state == Instance.States.UNAVAILABLE +@pytest.mark.django_db +def test_heartbeat_stops_when_this_instance_is_missing(settings): + """No record for this node and no auto-deprovision → returns (None, None, None, None).""" + settings.CLUSTER_HOST_ID = 'ctrl-0' + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + ctl = mock.MagicMock() + ctl.simple_command.return_value = {'KnownConnectionCosts': {'ctrl-0': {}}} + + with mock.patch('awx.main.tasks.system.get_receptor_ctl', return_value=ctl): + assert _heartbeat_instance_management() == (None, None, None, None) + + # ── AAP-89607: unified startup job processing ───────────────────────────────── diff --git a/awx/main/tests/unit/tasks/test_system.py b/awx/main/tests/unit/tasks/test_system.py index c567dc4833..a4c44d1f8d 100644 --- a/awx/main/tests/unit/tasks/test_system.py +++ b/awx/main/tests/unit/tasks/test_system.py @@ -1,6 +1,12 @@ import pytest +from datetime import datetime, timedelta, timezone from unittest.mock import MagicMock, patch -from awx.main.tasks.system import update_inventory_computed_fields +from awx.main.tasks.system import ( + _check_in_this_instance, + _receptor_mesh_status, + _register_missing_this_instance, + update_inventory_computed_fields, +) from awx.main.models import Inventory from django.db import DatabaseError @@ -62,3 +68,60 @@ def test_update_inventory_computed_fields_database_error_nosqlstate(mock_logger, mock_filter.assert_called_once_with(id=1) mock_update_computed_fields.assert_called_once() mock_inventory.update_computed_fields.assert_called_once() + + +NOW = datetime(2026, 1, 1, tzinfo=timezone.utc) + + +def test_receptor_mesh_status_returns_status(): + ctl = MagicMock() + ctl.simple_command.return_value = {'KnownConnectionCosts': {}} + + assert _receptor_mesh_status(ctl) == {'KnownConnectionCosts': {}} + + +def test_receptor_mesh_status_unavailable_is_none(mock_logger): + ctl = MagicMock() + ctl.simple_command.side_effect = OSError('socket gone') + + assert _receptor_mesh_status(ctl) is None + + +def _this_inst(lost=False, capacity=100, last_seen=NOW): + inst = MagicMock(hostname='ctrl-0', capacity=capacity, last_seen=last_seen) + inst.is_lost.return_value = lost + return inst + + +def test_check_in_rejoining_instance_returns_false(mock_logger): + inst = _this_inst(lost=True) + + assert _check_in_this_instance(inst, NOW) is False + inst.local_health_check.assert_called_once_with() + + +def test_check_in_lost_instance_without_capacity_continues(mock_logger): + assert _check_in_this_instance(_this_inst(lost=True, capacity=0), NOW) is True + + +@pytest.mark.parametrize('last_seen, warns', [(None, True), (NOW - timedelta(seconds=600), True), (NOW, False)]) +def test_check_in_warns_on_missing_or_skewed_last_seen(mock_logger, last_seen, warns): + assert _check_in_this_instance(_this_inst(last_seen=last_seen), NOW) is True + assert mock_logger.warning.called is warns + + +@pytest.mark.parametrize('changed', [True, False]) +def test_register_missing_this_instance_recreates_record(mock_logger, settings, changed): + settings.AWX_AUTO_DEPROVISION_INSTANCES = True + inst = MagicMock() + with patch('awx.main.tasks.system.Instance.objects.register', return_value=(changed, inst)): + assert _register_missing_this_instance() is inst + inst.local_health_check.assert_called_once_with() + assert mock_logger.warning.called is changed + + +def test_register_missing_this_instance_without_auto_deprovision(mock_logger, settings): + settings.AWX_AUTO_DEPROVISION_INSTANCES = False + + assert _register_missing_this_instance() is None + mock_logger.error.assert_called_once()