Initialize task app

Author: AlanCoding

ansible_base/task/__init__.py

@@ -0,0 +1,3 @@
+from django.contrib import admin
+
+# Register your models here.

ansible_base/task/admin.py

@@ -0,0 +1,8 @@
+from django.apps import AppConfig
+
+
+class TaskConfig(AppConfig):
+    default_auto_field = 'django.db.models.BigAutoField'
+    name = 'ansible_base.task'
+    label = 'dab_tas'
+    verbose_name = 'DAB tasking system'

ansible_base/task/apps.py

@@ -0,0 +1,26 @@
+from dispatcher.main import DispatcherMain
+
+from django.core.management.base import BaseCommand
+
+
+class Command(BaseCommand):
+    help = "Runs bug checking sanity checks, gets scale metrics, and recommendations for Role Based Access Control"
+
+    def handle(self, *args, **options):
+        dispatcher_config = {
+            "producers": {
+                "brokers": {
+                    "pg_notify": {"conninfo": settings.PG_NOTIFY_DSN_SERVER},
+                    # TODO: sanitize or escape channel names on dispatcher side
+                    "channels": [
+                        "dab_broadcast"
+                    ],
+                },
+                # NOTE: I would prefer to move the activation monitoring
+                # from worker to activation, but that is more work
+                "scheduled": {},
+            },
+            "pool": {"max_workers": 4},
+        }
+
+        DispatcherMain()

ansible_base/task/management/commands/run_dispatcher.py

@@ -0,0 +1,64 @@
+from uuid import uuid4
+from types import SimpleNamespace
+
+from django.db import models
+from django.utils.translation import gettext_lazy as _
+from django.utils.timezone import now
+
+# NOTE: tasks are not registered in the database, but in a dispatcher registry
+
+TASK_STATES = SimpleNamespace(
+    WAITING="waiting",
+    # SKIPPED="skipped",
+    RUNNING="running",
+    COMPLETED="completed",
+    FAILED="failed",
+    # CANCELED="canceled",
+    # CANCELING="canceling",
+)
+
+
+class Task(models.Model):
+    """
+    Corresponds to a call of a task, as a higher-level abstraction around the dispatcher.
+    Loosely modeled after pulpcore.Task
+    """
+    uuid = models.UUIDField(
+        primary_key=True, default=uuid4, editable=False,
+        help_text=_('UUID that corresponds to the dispatcher task uuid')
+    )
+    state = models.CharField(
+        choices=[(s, s.title()) for s in sorted(vars(TASK_STATES).values())],
+        default=TASK_STATES.WAITING,
+        help_text=_('Choices of this field track with acknowledgement and completion of a task')
+    )
+    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')
+    )
+    # pulp has unblocking logic, like unblocked_at, we have no plans for that here
+    started_at = models.DateTimeField(
+        null=True,
+        help_text=_('Time of acknowledgement, also approximately the time the task starts')
+    )
+    finished_at = models.DateTimeField(
+        null=True,
+        help_text=_('Time task is cleared (whether failed or succeeded), may be unused if set to auto-delete')
+    )
+
+    def mark_ack(self):
+        self.state = TASK_STATES.RUNNING
+        self.started_at = now()
+        self.save(update_fields=['state', 'started_at'])
+
+    def mark_completed(self):
+        self.state = TASK_STATES.COMPLETED
+        self.finished_at = now()
+
+    def mark_failed(self):
+        self.state = TASK_STATES.FAILED
+        self.finished_at = now()

ansible_base/task/migrations/__init__.py

@@ -0,0 +1,39 @@
+from django.db import transaction
+
+from dispatcher.utils import serialize_task
+
+from ansible_base.task.tasks import run_task_from_queue
+from ansible_base.task.models import Task
+
+# decorator structure is taken from dispatcher.publish
+
+class TaskPublisher:
+    def __init__(self, fn):
+        self.fn = fn
+
+    @property
+    def task_name(self):
+        return serialize_task(self.fn)
+
+    def apply_async(self, args=None, kwargs=None):
+        # this function may allow additional arguments in the future, but not now
+        Task.objects.create(name=self.task_name)
+
+        transaction.on_commit(run_task_from_queue.delay)
+
+    def delay(self, *args, **kwargs):
+        return self.apply_async(args=args, kwargs=kwargs)
+
+
+class TaskDecorator:
+    def __init__(self, *args, **kwargs):
+        self.args = args
+        self.kwargs = kwargs
+
+    def __call__(self, fn):
+        publisher = TaskPublisher(fn)
+
+        setattr(fn, 'apply_async', publisher.apply_async)
+        setattr(fn, 'delay', publisher.delay)
+
+        return fn

ansible_base/task/models.py

@@ -0,0 +1,30 @@
+import logging
+
+from ansible_base.task.models import Task, TASK_STATES
+
+from dispatcher.utils import resolve_callable
+
+from django.db import transaction
+
+logger = logging.getLogger(__name__)
+
+
+def run_task_from_queue():
+    with transaction.atomic():
+        task = Task.objects.filter(state=TASK_STATES.WAITING).select_for_update().first()
+        if task:
+            task.mark_ack()
+        else:
+            return
+
+    # for responsiveness with bursts of tasks
+    if Task.objects.filter(state=TASK_STATES.WAITING).exists():
+        run_task_from_queue.delay()
+
+    try:
+        task_callable = resolve_callable(task.name)
+        task_callable()
+    except Exception:
+        logger.traceback(f'Failed to run and complete {task.name}')
+
+    task.delete()

ansible_base/task/publish.py

@@ -0,0 +1,21 @@
+## DAB Task System Design
+
+The closest analog to the goals of this project may be the pulpcore task system
+
+https://github.com/pulp/pulpcore/tree/main/pulpcore/tasking
+
+The main loop is in the `worker` module under that. Like AWX, they use `select.select`,
+and combined with `add_notify_handler`.
+
+This accomplishes the goal of getting timely "wakeups" when a task is submitted.
+
+https://www.psycopg.org/psycopg3/docs/advanced/async.html
+
+> Alternatively, you can use add_notify_handler() to register a callback function, which will be invoked whenever a notification is received, during the normal query processing; you will be then able to use the connection normally. Please note that in this case notifications will not be received immediately, but only during a connection operation, such as a query.
+
+After control is returned, `unblock_tasks` loops over `Task` objects in order of created time.
+Based on the `Task` `status` field, it will take an action.
+
+Since uuid primary keys are standard in pulp, the default `pulp_id` field is a valid uuid for the task.
+
+While doing this, postgres advisory locks are used significantly to avoid errors from multiple simultaneous processes doing something.

ansible_base/task/tasks.py

@@ -0,0 +1,50 @@
+## DAB Task
+
+This app "dab_task" provides a tasking system (it can run tasks in the background).
+This is an overlay on top of the dispatcher library:
+
+https://github.com/ansible/dispatcher
+
+The key differences of using dab_task as opposed to the dispatcher directly are:
+
+ - Tasks are submitted to a true queue backed by postgres
+ - Formal message ACK de-duplicates submissions based on its UUID
+ - Rollbacks for database changes are possible, allowing for a number of retries
+
+The dispatcher library, itself, is intended to be lower-level,
+to accomidate cases where an app implements these things on its own.
+
+Obtaining these things come at a cost, mainly that almost all interactions
+require an additional database interaction.
+
+### Configuration
+
+Decorate tasks to allow them to be ran as background tasks.
+
+```python
+from ansible_base.dispatcher.publish import task
+
+@task()
+def hello_world():
+    print('hello world')
+```
+
+### Service
+
+To run the service, use the management command
+
+```
+python manage.py run_dispatcher
+```
+
+### Publisher
+
+Submit a task by importing it and using the interface from celery.
+
+```python
+from test_app.tasks import hello_world
+
+hello_world.delay()
+```
+
+This will print "hello world" in the dispatcher service.