Enable dispatcherd under feature flag 1st iteration - AAP-46009

.gitignore

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

pyproject.toml

@@ -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 = "*"

src/aap_eda/api/views/eda_credential.py

@@ -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/mixins.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/project.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/core/apps.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/dispatcherd_pre_fork.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/management/commands/rqworker.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,29 @@ def add_arguments(self, parser: CommandParser) -> None:
 
     def handle(self, *args, **options) -> None:
         if features.DISPATCHERD:
-            self.stderr.write(
+            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(
-                    "DISPATCHERD feature not implemented yet. "
-                    f"Please disable {settings.DISPATCHERD_FEATURE_FLAG_NAME} "
-                    "in your settings.",
+                    "Invalid worker class. "
+                    "Please use either ActivationWorker or DefaultWorker."
                 )
-            )
-            raise SystemExit(1)
+                raise SystemExit(1)
+
+            logger.info("Starting worker with dispatcherd.")
+            run_dispatcherd_service()
+            return None
+
+        # run rqworker command if dispatcherd is not enabled
+        logger.info("Starting worker with rqworker.")
         return rqworker.Command.handle(self, *args, **options)