Initialize task app

Move psycopg param generator to DAB utility

Store standard type annotation

Make a solution to getting conninfo as a string

Add another test and wrap up linters

Init migration, connect things

Add task container

Demo via a test_app view

Forgot procfile

Forgot to include tasks

Change name to durable_task

Get subprocess test to run

Get multi-server working

Adopt new bind=True kwarg

Pass through args and kwargs

Write task fallback logic

Move config into its own module, pop inside utils method

Basics of multi-queue working

Author: AlanCoding

Dockerfile

@@ -13,7 +13,8 @@ RUN dnf -y install \
     libtool-ltdl-devel \
     libpq-devel \
     libpq \
-    postgresql
+    postgresql \
+    git
 
 # Create /etc/ansible-automation-platform/testapp folder
 RUN mkdir -p /etc/ansible-automation-platform/testapp

Procfile

@@ -0,0 +1,2 @@
+web: python manage.py runserver
+task: python manage.py run_dispatcher

ansible_base/lib/dynamic_config/settings_logic.py

@@ -329,20 +329,17 @@ def get_mergeable_dab_settings(settings: dict) -> dict:  # NOSONAR
                 }
             )
 
-    # Set back only changed base keys so the inspect history is accurate
-    if installed_apps != settings["INSTALLED_APPS"]:
-        dab_data["INSTALLED_APPS"] = installed_apps
-    if middleware != settings["MIDDLEWARE"]:
-        dab_data["MIDDLEWARE"] = middleware
-    if rest_framework != settings["REST_FRAMEWORK"]:
-        dab_data["REST_FRAMEWORK"] = rest_framework
-    if oauth2_provider != settings.get("OAUTH2_PROVIDER", {}):
-        dab_data["OAUTH2_PROVIDER"] = oauth2_provider
-    if templates != settings.get("TEMPLATES", []):
-        dab_data["TEMPLATES"] = templates
-    if authentication_backends != settings.get("AUTHENTICATION_BACKENDS", []):
-        dab_data["AUTHENTICATION_BACKENDS"] = authentication_backends
-    if spectacular_settings != settings.get('SPECTACULAR_SETTINGS', {}):
-        dab_data['SPECTACULAR_SETTINGS'] = spectacular_settings
+    if 'ansible_base.task' in installed_apps:
+        # Queue used for dab_task tasks to re-schedule themselves if there is more work to do
+        # if you want to use a local queue instead, that would be more performant
+        dab_data['DAB_TASK_ADMIN_QUEUE'] = 'dab_broadcast'
+
+        # Queues that the local node will listen to
+        # useful to manage different pools of work and types of nodes
+        # this should probably include the admin queue
+        dab_data['DAB_TASK_LISTEN_QUEUES'] = ['dab_broadcast']
+
+        # Dispatcher spawns python subprocesses, so you can control the number of them here
+        dab_data['DAB_TASK_MAX_WORKERS'] = 4
 
     return dab_data

ansible_base/lib/utils/db.py

@@ -1,7 +1,12 @@
 from contextlib import contextmanager
+from copy import deepcopy
+from typing import Union
 from zlib import crc32
 
+import psycopg
+from django.conf import settings
 from django.db import DEFAULT_DB_ALIAS, connection, connections, transaction
+from django.db.backends.postgresql.base import DatabaseWrapper as PsycopgDatabaseWrapper
 from django.db.migrations.executor import MigrationExecutor
 
 
@@ -149,3 +154,62 @@ def advisory_lock(*args, lock_session_timeout_milliseconds=0, **kwargs):
         yield True
     else:
         raise RuntimeError(f'Advisory lock not implemented for database type {connection.vendor}')
+
+
+# Django settings.DATABASES['alias'] dictionary type
+dj_db_dict = dict[str, Union[str, int]]
+
+
+def psycopg_connection_from_django(**kwargs) -> psycopg.Connection:
+    "Compatibility with dispatcher connection factory, just returns the Django connection"
+    return connection.connection
+
+
+def psycopg_kwargs_from_settings_dict(settings_dict: dj_db_dict) -> dict:
+    """Return psycopg connection creation kwargs given Django db settings info
+
+    :param dict setting_dict: DATABASES in Django settings
+    :return: kwargs that can be passed to psycopg.connect, or connection classes"""
+    psycopg_params = PsycopgDatabaseWrapper(settings_dict).get_connection_params().copy()
+    psycopg_params.pop('cursor_factory', None)
+    psycopg_params.pop('context', None)
+    return psycopg_params
+
+
+def psycopg_conn_string_from_settings_dict(settings_dict: dj_db_dict) -> str:
+    conn_params = psycopg_kwargs_from_settings_dict(settings_dict)
+    return psycopg.conninfo.make_conninfo(**conn_params)
+
+
+def combine_settings_dict(settings_dict1: dj_db_dict, settings_dict2: dj_db_dict, **extra_options) -> dj_db_dict:
+    """Given two Django settings dictionaries, combine them and return a new settings_dict"""
+    settings_dict = deepcopy(settings_dict1)
+    settings_dict['OPTIONS'] = deepcopy(settings_dict.get('OPTIONS', {}))
+
+    # These extra options are used by AWX to set application_name
+    settings_dict['OPTIONS'].update(extra_options)
+
+    # Apply overrides specifically for the listener connection
+    for k, v in settings_dict2.items():
+        if k != 'OPTIONS':
+            settings_dict[k] = v
+
+    for k, v in settings_dict2.get('OPTIONS', {}).items():
+        settings_dict['OPTIONS'][k] = v
+
+    return settings_dict
+
+
+def get_pg_notify_params(alias: str = DEFAULT_DB_ALIAS, **extra_options) -> dict:
+    pg_notify_overrides = {}
+    if hasattr(settings, 'PG_NOTIFY_DATABASES'):
+        pg_notify_overrides = settings.PG_NOTIFY_DATABASES.get(alias, {})
+    elif hasattr(settings, 'LISTENER_DATABASES'):
+        pg_notify_overrides = settings.LISTENER_DATABASES.get(alias, {})
+
+    settings_dict = combine_settings_dict(settings.DATABASES[alias], pg_notify_overrides, **extra_options)
+
+    # Reuse the Django postgres DB backend to create params for the psycopg library
+    psycopg_params = psycopg_kwargs_from_settings_dict(settings_dict)
+
+    return psycopg_params

ansible_base/task/__init__.py

@@ -0,0 +1,15 @@
+from dispatcher.config import setup
+from django.apps import AppConfig
+
+from ansible_base.task.config import get_config
+
+
+class TaskConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'ansible_base.task'
+    label = 'dab_tas'
+    verbose_name = 'DAB tasking system'
+
+    def ready(self):
+        # Set up dispatcher
+        setup(get_config())

ansible_base/task/admin.py

@@ -0,0 +1,31 @@
+from django.conf import settings
+
+from ansible_base.lib.utils.db import get_pg_notify_params
+
+
+def get_config() -> dict:
+    "Returns dispatcher config from what is in Django settings"
+    psycopg_params = get_pg_notify_params()
+
+    return {
+        "version": 2,
+        "brokers": {
+            "pg_notify": {
+                "config": psycopg_params,  # used to create the service async connection
+                "sync_connection_factory": "ansible_base.lib.utils.db.psycopg_connection_from_django",
+                "channels": settings.DAB_TASK_LISTEN_QUEUES,
+                # The default publish channel allows using .delay without more arguments
+                # this may still have task-specific overrides
+                "default_publish_channel": settings.DAB_TASK_ADMIN_QUEUE,
+            }
+        },
+        "producers": {
+            "ScheduledProducer": {
+                "task_schedule": {
+                    "ansible_base.task.tasks.run_task_from_queue": {"schedule": 60},
+                    "ansible_base.task.tasks.manage_lost_tasks": {"schedule": 60 * 10},
+                }
+            },
+        },
+        "pool": {"max_workers": settings.DAB_TASK_MAX_WORKERS},
+    }

ansible_base/task/apps.py

@@ -0,0 +1,20 @@
+import logging
+
+from dispatcher import run_service
+from django.core.cache import cache
+from django.core.management.base import BaseCommand
+from django.db import connection
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    help = "Runs bug checking sanity checks, gets scale metrics, and recommendations for Role Based Access Control"
+
+    def handle(self, *args, **options):
+        # Borrowed from eda-server, ensure the database connection is closed
+        connection.close()
+        cache.close()
+
+        # Configuration is already handled in app .ready method
+        run_service()

ansible_base/task/config.py

@@ -0,0 +1,26 @@
+# Generated by Django 4.2.16 on 2025-02-04 21:22
+
+from django.db import migrations, models
+import uuid
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='Task',
+            fields=[
+                ('uuid', models.UUIDField(default=uuid.uuid4, editable=False, help_text='UUID that corresponds to the dispatcher task uuid', primary_key=True, serialize=False)),
+                ('state', models.CharField(choices=[('completed', 'Completed'), ('failed', 'Failed'), ('running', 'Running'), ('waiting', 'Waiting')], default='waiting', help_text='Choices of this field track with acknowledgement and completion of a task', max_length=15)),
+                ('name', models.TextField(help_text='Importable path for class or method')),
+                ('created', models.DateTimeField(auto_now_add=True, help_text='Time the publisher (submitter) of this task call created it, approximately the time of submission as well')),
+                ('started_at', models.DateTimeField(help_text='Time of acknowledgement, also approximately the time the task starts', null=True)),
+                ('finished_at', models.DateTimeField(help_text='Time task is cleared (whether failed or succeeded), may be unused if set to auto-delete', null=True)),
+            ],
+        ),
+    ]