fix: check if the lock was acquired and return if not (#16620)

* fix: check if the lock was acquired and return if not

* fix linter issues
This commit is contained in:
Peter Braun
2026-08-25 18:25:05 +02:00
committed by GitHub
parent d1fdad7718
commit 3917f3bf74
2 changed files with 77 additions and 2 deletions
+13 -1
View File
@@ -589,7 +589,11 @@ def inspect_established_receptor_connections(mesh_status):
def inspect_execution_and_hop_nodes(instance_list, receptor_ctl): def inspect_execution_and_hop_nodes(instance_list, receptor_ctl):
with advisory_lock('inspect_execution_and_hop_nodes_lock', wait=False): with advisory_lock('inspect_execution_and_hop_nodes_lock', wait=False) as acquired:
if not acquired:
logger.debug("Not running inspect_execution_and_hop_nodes, another instance holds lock")
return
start = time.monotonic()
node_lookup = {inst.hostname: inst for inst in instance_list} node_lookup = {inst.hostname: inst for inst in instance_list}
try: try:
mesh_status = receptor_ctl.simple_command('status') mesh_status = receptor_ctl.simple_command('status')
@@ -601,6 +605,7 @@ def inspect_execution_and_hop_nodes(instance_list, receptor_ctl):
nowtime = now() nowtime = now()
workers = mesh_status['Advertisements'] workers = mesh_status['Advertisements']
updated_count = 0
for ad in workers: for ad in workers:
hostname = ad['NodeID'] hostname = ad['NodeID']
@@ -620,6 +625,7 @@ def inspect_execution_and_hop_nodes(instance_list, receptor_ctl):
continue continue
instance.last_seen = last_seen instance.last_seen = last_seen
instance.save(update_fields=['last_seen']) instance.save(update_fields=['last_seen'])
updated_count += 1
# Only execution nodes should be dealt with by execution_node_health_check # Only execution nodes should be dealt with by execution_node_health_check
if instance.node_type == Instance.Types.HOP: if instance.node_type == Instance.Types.HOP:
@@ -642,6 +648,12 @@ def inspect_execution_and_hop_nodes(instance_list, receptor_ctl):
logger.debug(f'Restarting health check for execution node {hostname} with known errors.') logger.debug(f'Restarting health check for execution node {hostname} with known errors.')
execution_node_health_check.apply_async([hostname]) execution_node_health_check.apply_async([hostname])
elapsed = time.monotonic() - start
if elapsed > 2.0:
logger.warning(f"inspect_execution_and_hop_nodes completed in {elapsed:.1f}s, updated {updated_count} node(s)")
else:
logger.debug(f"inspect_execution_and_hop_nodes completed in {elapsed:.3f}s, updated {updated_count} node(s)")
@task(queue=get_task_queuename, bind=True) @task(queue=get_task_queuename, bind=True)
def cluster_node_heartbeat(binder): def cluster_node_heartbeat(binder):
+64 -1
View File
@@ -1,6 +1,6 @@
import pytest import pytest
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from awx.main.tasks.system import _heartbeat_instance_management, update_inventory_computed_fields from awx.main.tasks.system import _heartbeat_instance_management, update_inventory_computed_fields, inspect_execution_and_hop_nodes
from awx.main.models import Instance, Inventory from awx.main.models import Instance, Inventory
from django.db import DatabaseError from django.db import DatabaseError
@@ -78,3 +78,66 @@ def test_heartbeat_marks_offline_when_receptor_unavailable(mock_filter, mock_get
assert result == (None, None, None) assert result == (None, None, None)
this_inst.local_health_check.assert_called_once() this_inst.local_health_check.assert_called_once()
this_inst.mark_offline.assert_called_once_with(errors='Receptor not available') this_inst.mark_offline.assert_called_once_with(errors='Receptor not available')
class TestInspectExecutionAndHopNodes:
"""Tests for the advisory_lock guard in inspect_execution_and_hop_nodes."""
@patch('awx.main.tasks.system.inspect_established_receptor_connections')
@patch('awx.main.tasks.system.advisory_lock')
def test_skips_when_lock_not_acquired(self, mock_lock, mock_inspect_conns):
mock_lock.return_value.__enter__ = MagicMock(return_value=False)
mock_lock.return_value.__exit__ = MagicMock(return_value=False)
receptor_ctl = MagicMock()
inspect_execution_and_hop_nodes([], receptor_ctl)
receptor_ctl.simple_command.assert_not_called()
mock_inspect_conns.assert_not_called()
@patch('awx.main.tasks.system.inspect_established_receptor_connections')
@patch('awx.main.tasks.system.advisory_lock')
def test_runs_when_lock_acquired(self, mock_lock, mock_inspect_conns):
mock_lock.return_value.__enter__ = MagicMock(return_value=True)
mock_lock.return_value.__exit__ = MagicMock(return_value=False)
receptor_ctl = MagicMock()
receptor_ctl.simple_command.return_value = {'Advertisements': [], 'KnownConnectionCosts': {}}
inspect_execution_and_hop_nodes([], receptor_ctl)
receptor_ctl.simple_command.assert_called_once_with('status')
mock_inspect_conns.assert_called_once()
@patch('awx.main.tasks.system.inspect_established_receptor_connections')
@patch('awx.main.tasks.system.advisory_lock')
def test_updates_last_seen_for_execution_nodes(self, mock_lock, mock_inspect_conns):
mock_lock.return_value.__enter__ = MagicMock(return_value=True)
mock_lock.return_value.__exit__ = MagicMock(return_value=False)
exec_node = MagicMock(spec=Instance)
exec_node.hostname = 'exec-1'
exec_node.node_type = 'execution'
exec_node.node_state = 'ready'
exec_node.last_seen = None
exec_node.capacity = 100
exec_node.enabled = True
exec_node.cpu = 4
exec_node.memory = 8000000000
control_node = MagicMock(spec=Instance)
control_node.hostname = 'control-1'
control_node.node_type = 'control'
receptor_ctl = MagicMock()
receptor_ctl.simple_command.return_value = {
'Advertisements': [
{'NodeID': 'exec-1', 'Time': '2026-01-01T00:00:00+00:00'},
{'NodeID': 'control-1', 'Time': '2026-01-01T00:00:00+00:00'},
],
'KnownConnectionCosts': {},
}
inspect_execution_and_hop_nodes([exec_node, control_node], receptor_ctl)
exec_node.save.assert_called_once_with(update_fields=['last_seen'])
control_node.save.assert_not_called()