Fix Task Decorator to Work With and Without Feature Flag (AAP-41775)

[art-tapin]

SUMMARY

Implemented a solution that enables AWX's task system to work with both the legacy dispatcher and the new dispatcherd library based on the FEATURE_NEW_DISPATCHER feature flag. Created a registry pattern instead of trying to modify method objects directly.

ISSUE TYPE

• Bug Fix

COMPONENT NAME

• Other (Dispatcher System)

AWX VERSION

ADDITIONAL INFORMATION

Problem:
When initially tried to modify methods directly with cluster_node_heartbeat.apply_async._new_method = ..., I encountered AttributeError: 'method' object has no attribute '_new_method' because Python bound methods don't support attribute assignment.

With the feature flag disabled, I also encountered this scheduler error (probably known):

File "/awx_devel/awx/main/dispatch/periodic.py", line 26, in __init__
  self.interval = int(data['schedule'].total_seconds())
AttributeError: 'float' object has no attribute 'total_seconds'

Solution:

1. In publish.py:

   • Added ALTERNATIVE_TASK_IMPLEMENTATIONS registry at module level
   • Modified apply_async to check the registry when the feature flag is enabled

2. In system.py:

   • Refactored cluster_node_heartbeat into two implementations:
     • Original AWX version with bind_kwargs
     • New dispatcherd version using bind=True
   • Extracted common code into helper functions
   • Added _get_active_task_ids_from_dispatcherd() to retrieve running tasks from the new dispatcher
   • Registered the alternative implementation in the central registry

3. Other minor adjustments to ensure compatibility with both systems

Testing Steps:

# (For ansible org) Pull the right image
docker image pull ghcr.io/ansible/awx_devel:feature_dispatcher

# Start AWX with feature dispatcher
COMPOSE_TAG=feature_dispatcher make docker-compose

# In AWX container, test with flag enabled:
# First, enable it in awx/settings/development_defaults.py
awx-manage shell -c "from awx.main.tasks.system import cluster_node_heartbeat; cluster_node_heartbeat.delay()"
# From the logs
grep "Using dispatcherd implementation"

# Test with flag disabled:
# First, enable it in awx/settings/development_defaults.py
awx-manage shell -c "from awx.main.tasks.system import cluster_node_heartbeat; cluster_node_heartbeat.delay()"

For more extensive in-the-code debugging logs, checkout the revision (commit): fix(dispatcher): Implement registry pattern for dispatcher feature flag compatibility

[AlanCoding]

AttributeError: 'float' object has no attribute 'total_seconds'

yeah, switched from using a timedelta to a float for seconds, and this is fallout from that.

EDIT: Oh now I know what the issue is. It was in-place modifications I had done

diff --git a/awx/settings/defaults.py b/awx/settings/defaults.py
index 43cbab180a..d2b242eadf 100644
--- a/awx/settings/defaults.py
+++ b/awx/settings/defaults.py
@@ -456,7 +456,7 @@ CELERYBEAT_SCHEDULE = {
 DISPATCHER_SCHEDULE = {}
 for options in CELERYBEAT_SCHEDULE.values():
     task_name = options['task']
-    DISPATCHER_SCHEDULE[task_name] = options
+    DISPATCHER_SCHEDULE[task_name] = options.copy()
     DISPATCHER_SCHEDULE[task_name]['schedule'] = options['schedule'].total_seconds()
 
 # Django Caching Configuration

[AlanCoding]

With the 1 modification for scheduled jobs, able to run the Demo Job Template and schedules run.

[art-tapin]

@AlanCoding I've added the .copy() method to create a proper copy in this PR, if you're okay with it

AttributeError: 'float' object has no attribute 'total_seconds'

yeah, switched from using a timedelta to a float for seconds, and this is fallout from that.

EDIT: Oh now I know what the issue is. It was in-place modifications I had done

diff --git a/awx/settings/defaults.py b/awx/settings/defaults.py
index 43cbab180a..d2b242eadf 100644
--- a/awx/settings/defaults.py
+++ b/awx/settings/defaults.py
@@ -456,7 +456,7 @@ CELERYBEAT_SCHEDULE = {
 DISPATCHER_SCHEDULE = {}
 for options in CELERYBEAT_SCHEDULE.values():
     task_name = options['task']
-    DISPATCHER_SCHEDULE[task_name] = options
+    DISPATCHER_SCHEDULE[task_name] = options.copy()
     DISPATCHER_SCHEDULE[task_name]['schedule'] = options['schedule'].total_seconds()
 
 # Django Caching Configuration

[sonarqubecloud]

Quality Gate passed

Issues
5 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
0.0% Duplication on New Code

See analysis details on SonarQube Cloud