Enable dispatcherd under feature flag 1st iteration - AAP-46009 (#1308)

Initial minimal implementation of dispatcherd under the feature flag.
Current rq deployment tested
Still missing tests for dispatcherd and proxies that we are going to
address in later iterations as part of
https://issues.redhat.com/browse/AAP-46008

Dispatcherd is widely tested as well but not guaranteed a this point. 
~WE NEED TO CONFIRM THE AVAILABILITY OF THE RPM PRIOR TO MERGE THIS~,
(tracked internally in https://issues.redhat.com/browse/AAP-45706)
Update: RPM done. 

Jira: https://issues.redhat.com/browse/AAP-46009

---------

Signed-off-by: Alex <[email protected]>
Co-authored-by: Bill Wei <[email protected]>

Author: Alex-Izquierdo

.gitignore

@@ -27,4 +27,7 @@ Taskfile.yaml
 venv/
 .coverage*
 coverage.xml
-test-results.xml
+test-results.xml
+
+# Dispatcherd development
+dispatcherd/

poetry.lock

@@ -47,7 +47,7 @@ cryptography = ">=42,<43"
 kubernetes = "26.1.*"
 podman = "5.4.*"
 rq-scheduler = "^0.10"
-django-ansible-base = { git = "https://github.com/ansible/django-ansible-base.git", tag = "2025.3.7", extras = [
+django-ansible-base = { git = "https://github.com/ansible/django-ansible-base.git", tag = "2025.5.8", extras = [
     "channel-auth",
     "rbac",
     "redis-client",
@@ -70,6 +70,8 @@ validators = "^0.34.0"
 django-flags = "^5.0.13"
 insights-analytics-collector = "^0.3.2"
 distro = "^1.9.0"
+dispatcherd = { version = "v2025.05.19", extras = ["pg_notify"] }
+
 
 [tool.poetry.group.test.dependencies]
 pytest = "*"

pyproject.toml

@@ -214,7 +214,7 @@ def partial_update(self, request, pk):
         eda_credential.refresh_from_db()
         new_interval = get_analytics_interval_if_exist(eda_credential)
         if new_interval != old_interval:
-            reschedule_gather_analytics.delay()
+            reschedule_gather_analytics()
         return Response(
             serializers.EdaCredentialSerializer(eda_credential).data,
         )
@@ -307,7 +307,7 @@ def _create_eda_credential(self, request, data):
             )
 
         if get_analytics_interval_if_exist(response) > 0:
-            reschedule_gather_analytics.delay()
+            reschedule_gather_analytics()
 
         return Response(
             serializers.EdaCredentialSerializer(response).data,

src/aap_eda/api/views/eda_credential.py

@@ -26,6 +26,7 @@
 
 from aap_eda.api import exceptions as api_exc
 from aap_eda.core import tasking
+from aap_eda.settings import features
 
 
 # TODO: need revisit from cuwater
@@ -147,5 +148,8 @@ def redis_is_available(
         self,
         message: Optional[str] = "Redis is required but unavailable.",
     ):
+        # dispatcherd does not need redis
+        if features.DISPATCHERD:
+            return True
         if tasking.is_redis_failed():
             raise api_exc.Conflict(message)

src/aap_eda/api/views/mixins.py

@@ -135,15 +135,14 @@ def create(self, request):
                     request.user, serializer.instance
                 )
 
-                job = tasks.import_project.delay(project_id=project.id)
+                job_id = tasks.import_project(project.id)
         except redis.ConnectionError:
             return RedisDependencyMixin.redis_unavailable_response()
 
-        # Atomically update `import_task_id` field only.
         models.Project.objects.filter(pk=project.id).update(
-            import_task_id=job.id
+            import_task_id=job_id
         )
-        project.import_task_id = job.id
+        project.import_task_id = job_id
         serializer = self.get_serializer(project)
         headers = self.get_success_headers(serializer.data)
         logger.info(
@@ -271,16 +270,22 @@ def sync(self, request, pk):
                 detail="Project import or sync is already running."
             )
 
+        # check if redis is available
         self.redis_is_available()
 
         try:
-            job = tasks.sync_project.delay(project_id=project.id)
+            job_id = tasks.sync_project(project.id)
         except redis.ConnectionError:
             return RedisDependencyMixin.redis_unavailable_response()
 
         project.import_state = models.Project.ImportState.PENDING
-        project.import_task_id = job.id
-        project.import_error = None
+
+        # job_id can be none if there is already a task running.
+        # this is unlikely since we check the state above
+        # but safety first
+        if job_id:
+            project.import_task_id = job_id
+
         project.save()
 
         logger.info(

src/aap_eda/api/views/project.py

@@ -15,7 +15,9 @@
 import logging
 import sys
 
+from dispatcherd.config import setup as dispatcher_setup
 from django.apps import AppConfig
+from django.conf import settings
 
 logger = logging.getLogger(__name__)
 
@@ -29,8 +31,11 @@ def ready(self):
         from aap_eda.api.views import dab_decorate  # noqa: F401
 
         # Run the startup logging for rq worker
-
+        # WARNING: rqworker can run rq workers or dispatcherd workers
         if "rqworker" in sys.argv:
             from aap_eda.utils.logging import startup_logging
 
             startup_logging(logger)
+
+        # Enable default dispatcher config. Workers may override this
+        dispatcher_setup(settings.DISPATCHERD_DEFAULT_SETTINGS)

src/aap_eda/core/apps.py

@@ -0,0 +1,34 @@
+#  Copyright 2025 Red Hat, Inc.
+#
+#  Licensed under the Apache License, Version 2.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+#  See the License for the specific language governing permissions and
+#  limitations under the License.
+
+import django
+from django.core.cache import cache
+from django.db import connection
+
+"""This module is an optimization for dispatcherd workers
+
+This sets up Django pre-fork, which must be implemented as a module to run
+on-import for compatibility with multiprocessing forkserver.
+This should never be imported by other modules, which is why it is called
+hazmat.
+"""
+
+
+django.setup()
+
+# connections may or may not be open, but
+# before forking, all connections should be closed
+
+cache.close()
+connection.close()

src/aap_eda/core/dispatcherd_pre_fork.py

@@ -14,12 +14,18 @@
 
 """Wrapper for rqworker command."""
 
+import logging
+
+from dispatcherd import run_service as run_dispatcherd_service
+from dispatcherd.config import setup as dispatcherd_setup
 from django.conf import settings
 from django.core.management.base import BaseCommand, CommandParser
 from django_rq.management.commands import rqworker
 
 from aap_eda.settings import features
 
+logger = logging.getLogger(__name__)
+
 
 class Command(BaseCommand):
     """Wrapper for rqworker command.
@@ -35,12 +41,33 @@ def add_arguments(self, parser: CommandParser) -> None:
 
     def handle(self, *args, **options) -> None:
         if features.DISPATCHERD:
-            self.stderr.write(
-                self.style.ERROR(
-                    "DISPATCHERD feature not implemented yet. "
-                    f"Please disable {settings.DISPATCHERD_FEATURE_FLAG_NAME} "
-                    "in your settings.",
-                )
+            return self._handle_dispatcherd(*args, **options)
+
+        # run rqworker command if dispatcherd is not enabled
+        logger.info("Starting worker with rqworker.")
+        return rqworker.Command.handle(self, *args, **options)
+
+    def _handle_dispatcherd(self, *args, **options) -> None:
+        """Handle dispatcherd service."""
+        if "worker_class" not in options:
+            self.style.ERROR("Missing required argument: --worker-class")
+            raise SystemExit(1)
+
+        # Use rqworker expected args to determine worker type
+        if "ActivationWorker" in options["worker_class"]:
+            dispatcherd_setup(
+                settings.DISPATCHERD_ACTIVATION_WORKER_SETTINGS,
+            )
+
+        elif "DefaultWorker" in options["worker_class"]:
+            dispatcherd_setup(settings.DISPATCHERD_DEFAULT_WORKER_SETTINGS)
+        else:
+            self.style.ERROR(
+                "Invalid worker class. "
+                "Please use either ActivationWorker or DefaultWorker."
             )
             raise SystemExit(1)
-        return rqworker.Command.handle(self, *args, **options)
+
+        logger.info("Starting worker with dispatcherd.")
+        run_dispatcherd_service()
+        return None