Modifying reaper of administrative work units to allow for change from Controller to Hybrid nodes (#12614)

awx/main/tasks/receptor.py

@@ -99,16 +99,22 @@ def administrative_workunit_reaper(work_list=None):
 
     for unit_id, work_data in work_list.items():
         extra_data = work_data.get('ExtraData')
-        if (extra_data is None) or (extra_data.get('RemoteWorkType') != 'ansible-runner'):
+        if extra_data is None:
             continue  # if this is not ansible-runner work, we do not want to touch it
-        params = extra_data.get('RemoteParams', {}).get('params')
+        if isinstance(extra_data, str):
-        if not params:
+            if not work_data.get('StateName', None) or work_data.get('StateName') in RECEPTOR_ACTIVE_STATES:
-            continue
+                continue
-        if not (params == '--worker-info' or params.startswith('cleanup')):
+        else:
-            continue  # if this is not a cleanup or health check, we do not want to touch it
+            if extra_data.get('RemoteWorkType') != 'ansible-runner':
-        if work_data.get('StateName') in RECEPTOR_ACTIVE_STATES:
+                continue
-            continue  # do not want to touch active work units
+            params = extra_data.get('RemoteParams', {}).get('params')
-        logger.info(f'Reaping orphaned work unit {unit_id} with params {params}')
+            if not params:
+                continue
+            if not (params == '--worker-info' or params.startswith('cleanup')):
+                continue  # if this is not a cleanup or health check, we do not want to touch it
+            if work_data.get('StateName') in RECEPTOR_ACTIVE_STATES:
+                continue  # do not want to touch active work units
+            logger.info(f'Reaping orphaned work unit {unit_id} with params {params}')
         receptor_ctl.simple_command(f"work release {unit_id}")
 
 

awx/main/tests/unit/test_tasks.py

@@ -34,7 +34,7 @@ from awx.main.models import (
 )
 from awx.main.models.credential import HIDDEN_PASSWORD, ManagedCredentialType
 
-from awx.main.tasks import jobs, system
+from awx.main.tasks import jobs, system, receptor
 from awx.main.utils import encrypt_field, encrypt_value
 from awx.main.utils.safe_yaml import SafeLoader
 from awx.main.utils.execution_environments import CONTAINER_ROOT
@@ -42,6 +42,8 @@ from awx.main.utils.execution_environments import CONTAINER_ROOT
 from awx.main.utils.licensing import Licenser
 from awx.main.constants import JOB_VARIABLE_PREFIXES
 
+from receptorctl.socket_interface import ReceptorControl
+
 
 def to_host_path(path, private_data_dir):
     """Given a path inside of the EE container, this gives the absolute path
@@ -1965,3 +1967,120 @@ def test_project_update_no_ee(mock_me):
         task.build_env(job, {})
 
     assert 'The project could not sync because there is no Execution Environment' in str(e.value)
+
+
+@pytest.mark.parametrize(
+    'work_unit_data, expected_function_call',
+    [
+        [
+            # if (extra_data is None): continue
+            {
+                'zpdFi4BX': {
+                    'ExtraData': None,
+                }
+            },
+            False,
+        ],
+        [
+            # Extra data is a string and StateName is None
+            {
+                "y4NgMKKW": {
+                    "ExtraData": "Unknown WorkType",
+                }
+            },
+            False,
+        ],
+        [
+            # Extra data is a string and StateName in RECEPTOR_ACTIVE_STATES
+            {
+                "y4NgMKKW": {
+                    "ExtraData": "Unknown WorkType",
+                    "StateName": "Running",
+                }
+            },
+            False,
+        ],
+        [
+            # Extra data is a string and StateName not in RECEPTOR_ACTIVE_STATES
+            {
+                "y4NgMKKW": {
+                    "ExtraData": "Unknown WorkType",
+                    "StateName": "Succeeded",
+                }
+            },
+            True,
+        ],
+        [
+            # Extra data is a dict but RemoteWorkType is not ansible-runner
+            {
+                "y4NgMKKW": {
+                    'ExtraData': {
+                        'RemoteWorkType': 'not-ansible-runner',
+                    },
+                }
+            },
+            False,
+        ],
+        [
+            # Extra data is a dict and its an ansible-runner but we have no params
+            {
+                'zpdFi4BX': {
+                    'ExtraData': {
+                        'RemoteWorkType': 'ansible-runner',
+                    },
+                }
+            },
+            False,
+        ],
+        [
+            # Extra data is a dict and its an ansible-runner but params is not --worker-info
+            {
+                'zpdFi4BX': {
+                    'ExtraData': {'RemoteWorkType': 'ansible-runner', 'RemoteParams': {'params': '--not-worker-info'}},
+                }
+            },
+            False,
+        ],
+        [
+            # Extra data is a dict and its an ansible-runner but params starts without cleanup
+            {
+                'zpdFi4BX': {
+                    'ExtraData': {'RemoteWorkType': 'ansible-runner', 'RemoteParams': {'params': 'not cleanup stuff'}},
+                }
+            },
+            False,
+        ],
+        [
+            # Extra data is a dict and its an ansible-runner w/ params but still running
+            {
+                'zpdFi4BX': {
+                    'ExtraData': {'RemoteWorkType': 'ansible-runner', 'RemoteParams': {'params': '--worker-info'}},
+                    "StateName": "Running",
+                }
+            },
+            False,
+        ],
+        [
+            # Extra data is a dict and its an ansible-runner w/ params and completed
+            {
+                'zpdFi4BX': {
+                    'ExtraData': {'RemoteWorkType': 'ansible-runner', 'RemoteParams': {'params': '--worker-info'}},
+                    "StateName": "Succeeded",
+                }
+            },
+            True,
+        ],
+    ],
+)
+def test_administrative_workunit_reaper(work_unit_data, expected_function_call):
+    # Mock the get_receptor_ctl call and let it return a dummy object
+    # It does not matter what file name we return as the socket because we won't actually call receptor (unless something is broken)
+    with mock.patch('awx.main.tasks.receptor.get_receptor_ctl') as mock_get_receptor_ctl:
+        mock_get_receptor_ctl.return_value = ReceptorControl('/var/run/awx-receptor/receptor.sock')
+        with mock.patch('receptorctl.socket_interface.ReceptorControl.simple_command') as simple_command:
+            receptor.administrative_workunit_reaper(work_list=work_unit_data)
+
+    if expected_function_call:
+        simple_command.assert_called()
+    else:
+        simple_command.assert_not_called()