Use django-q2 for notifications

Dockerfile

@@ -19,4 +19,4 @@ ENV PORT=8000
 EXPOSE 8000
 
 ENTRYPOINT ["/usr/bin/tini", "-v", "--"]
-CMD ["/argus/docker-entrypoint.sh"]
+CMD ["/argus/docker-entrypoint-api.sh"]

docker-compose.yml

@@ -1,26 +1,33 @@
 version: '3.5'
+
+x-api: &api-service
+  build: .
+  depends_on:
+    - postgres
+    - redis
+  environment:
+    - TIME_ZONE=Europe/Oslo
+    - DJANGO_SETTINGS_MODULE=argus.site.settings.dockerdev
+    - DATABASE_URL=postgresql://argus:HahF9araeKoo@postgres/argus
+    - ARGUS_REDIS_SERVER=redis
+  volumes:
+    - ${PWD}:/argus
+  restart: always
+
 services:
   api:
-    build: .
+    <<: *api-service
     ports:
       - "8000:8000"
-    depends_on:
-      - postgres
-      - redis
-    environment:
-      - TIME_ZONE=Europe/Oslo
-      - DJANGO_SETTINGS_MODULE=argus.site.settings.dockerdev
-      - DATABASE_URL=postgresql://argus:HahF9araeKoo@postgres/argus
-      - ARGUS_REDIS_SERVER=redis
-    volumes:
-      - ${PWD}:/argus
-    restart: always
 
+  qcluster:
+    <<: *api-service
+    command: ./docker-entrypoint-qcluster.sh
 
   postgres:
-    image: "postgres:12"
+    image: "postgres:13"
     volumes:
-      - postgres:/var/lib/postgresql/data:Z
+      - postgres:/var/lib/postgresql/data
     restart: always
     environment:
       - POSTGRES_USER=argus

docker-entrypoint-qcluster.sh

@@ -0,0 +1,5 @@
+#!/bin/bash -xe
+
+cd /argus
+python3 -m pip install -e .
+exec python3 manage.py qcluster

docker/Dockerfile

@@ -1,6 +1,6 @@
 # Defines a production image for the Argus API server
 # Needs the repository root directory as its context
-FROM python:3.9
+FROM python:3.11
 ENV DEBIAN_FRONTEND=noninteractive
 
 RUN apt-get update && apt-get install -y --no-install-recommends tini build-essential
@@ -19,5 +19,6 @@ ENV DJANGO_SETTINGS_MODULE=dockersettings
 
 ENV PORT=8000
 EXPOSE 8000
-COPY docker/docker-entrypoint.sh /api-entrypoint.sh
-ENTRYPOINT ["/usr/bin/tini", "-v", "--", "/api-entrypoint.sh"]
+COPY docker/docker-entrypoint-api.sh /api-entrypoint-api.sh
+COPY docker/docker-entrypoint-qcluster.sh /api-entrypoint-qcluster.sh
+ENTRYPOINT ["/usr/bin/tini", "-v", "--", "/api-entrypoint-api.sh"]

docker/README.md

@@ -18,7 +18,19 @@ Or, from the top level directory:
 docker build -t argus -f docker/Dockerfile .
 ```
 
-## Configuration of the running container
+## Services
+
+While this image provides the necessary environment for the backend processes,
+it defaults to run only the API server itself.  In addition to an API
+container, a second container is needed to process notifications
+asynchronously.  The second container needs to run from this same image, but
+you should replace the container command with `django-admin qcluster`, to run
+the "qcluster" service.
+
+The "qcluster" command comes from Django Q2, and is used to process a message
+queue of tasks to complete in the background.
+
+## Configuration of the running containers
 
 This image runs with default production settings, with a few tweaks from
 [dockersettings.py](dockersettings.py). This means that the most useful

pyproject.toml

@@ -31,6 +31,7 @@ dependencies = [
     "django-phonenumber-field[phonenumberslite]",
     "djangorestframework>=3.14",
     "drf-rw-serializers>=1.1",
+    "django-q2",
     "drf-spectacular>=0.17",
     "factory_boy",
     "psycopg2",

requirements-django42.txt

@@ -70,6 +70,8 @@ django==4.2.11
     #   django-cors-headers
     #   django-filter
     #   django-phonenumber-field
+    #   django-picklefield
+    #   django-q2
     #   djangorestframework
     #   drf-rw-serializers
     #   drf-spectacular
@@ -80,6 +82,10 @@ django-filter==23.2
     # via argus-server (pyproject.toml)
 django-phonenumber-field[phonenumberslite]==7.1.0
     # via argus-server (pyproject.toml)
+django-picklefield==3.1
+    # via django-q2
+django-q2==1.5.2
+    # via argus-server (pyproject.toml)
 djangorestframework==3.14.0
     # via
     #   argus-server (pyproject.toml)

requirements-django50.txt

@@ -72,6 +72,8 @@ django==5.0.2
     #   django-filter
     #   django-multiselectfield
     #   django-phonenumber-field
+    #   django-picklefield
+    #   django-q2
     #   djangorestframework
     #   drf-rw-serializers
     #   drf-spectacular
@@ -84,6 +86,10 @@ django-multiselectfield==0.1.12
     # via argus-server (pyproject.toml)
 django-phonenumber-field[phonenumberslite]==7.3.0
     # via argus-server (pyproject.toml)
+django-picklefield==3.1
+    # via django-q2
+django-q2==1.6.1
+    # via argus-server (pyproject.toml)
 djangorestframework==3.14.0
     # via
     #   argus-server (pyproject.toml)

requirements.txt

@@ -70,6 +70,8 @@ django==4.2.11
     #   django-cors-headers
     #   django-filter
     #   django-phonenumber-field
+    #   django-picklefield
+    #   django-q2
     #   djangorestframework
     #   drf-rw-serializers
     #   drf-spectacular
@@ -80,6 +82,10 @@ django-filter==23.2
     # via argus-server (pyproject.toml)
 django-phonenumber-field[phonenumberslite]==7.1.0
     # via argus-server (pyproject.toml)
+django-picklefield==3.1
+    # via django-q2
+django-q2==1.5.2
+    # via argus-server (pyproject.toml)
 djangorestframework==3.14.0
     # via
     #   argus-server (pyproject.toml)

src/argus/incident/apps.py

@@ -11,12 +11,11 @@ def ready(self):
             close_token_incident,
             delete_associated_user,
             delete_associated_event,
-            task_send_notification,  # noqa
-            task_background_send_notification,
+            enqueue_event_for_notification,
         )
 
         post_delete.connect(delete_associated_user, "argus_incident.SourceSystem")
         post_delete.connect(delete_associated_event, "argus_incident.Acknowledgement")
         post_delete.connect(close_token_incident, "authtoken.Token")
         post_save.connect(close_token_incident, "authtoken.Token")
-        post_save.connect(task_background_send_notification, "argus_incident.Event", dispatch_uid="send_notification")
+        post_save.connect(enqueue_event_for_notification, "argus_incident.Event", dispatch_uid="send_notification")

src/argus/incident/signals.py

@@ -1,35 +1,52 @@
+import logging
+
 from django.db.models import Q
 from django.utils import timezone
+
 from rest_framework.authtoken.models import Token
+from django_q.tasks import async_task
 
-from argus.notificationprofile.media import send_notifications_to_users
-from argus.notificationprofile.media import background_send_notification
+from argus.notificationprofile.media import find_destinations_for_event
 from argus.notificationprofile.media import send_notification
+from argus.notificationprofile.media import are_notifications_turned_on
+
 from .models import (
-    Acknowledgement, ChangeEvent, Event, Incident, SourceSystem, Tag,
+    Acknowledgement,
+    ChangeEvent,
+    Event,
+    Incident,
+    SourceSystem,
+    Tag,
     get_or_create_default_instances,
 )
 
 
 __all__ = [
+    "enqueue_event_for_notification",
     "delete_associated_user",
     "send_notification",
     "delete_associated_event",
     "close_token_incident",
 ]
 
+LOG = logging.getLogger(__name__)
 
-def delete_associated_user(sender, instance: SourceSystem, *args, **kwargs):
-    if hasattr(instance, "user") and instance.user:
-        instance.user.delete()
 
+def enqueue_event_for_notification(sender, instance: Event, *args, **kwargs):
+    if not are_notifications_turned_on():
+        return
 
-def task_send_notification(sender, instance: Event, *args, **kwargs):
-    send_notifications_to_users(instance)
+    destinations = find_destinations_for_event(instance)
+    if destinations:
+        LOG.info('Notification: will be sending notification for "%s"', instance)
+        async_task(send_notification, destinations, instance, group="notifications")
+    else:
+        LOG.debug("Notification: no destinations to send notification to")
 
 
-def task_background_send_notification(sender, instance: Event, *args, **kwargs):
-    send_notifications_to_users(instance, send=background_send_notification)
+def delete_associated_user(sender, instance: SourceSystem, *args, **kwargs):
+    if hasattr(instance, "user") and instance.user:
+        instance.user.delete()
 
 
 def delete_associated_event(sender, instance: Acknowledgement, *args, **kwargs):

src/argus/incident/views.py

@@ -28,10 +28,6 @@
     TicketPluginException,
     TicketSettingsException,
 )
-from argus.notificationprofile.media import (
-    send_notifications_to_users,
-    background_send_notification,
-)
 from argus.util.datetime_utils import INFINITY_REPR
 from argus.util.signals import bulk_changed
 from argus.util.utils import import_class_from_dotted_path

src/argus/notificationprofile/media/__init__.py

@@ -1,11 +1,9 @@
 from __future__ import annotations
 
 import logging
-from multiprocessing import Process
 from typing import TYPE_CHECKING
 
 from django.conf import settings
-from django.db import connections
 from rest_framework.exceptions import ValidationError
 
 from ..models import DestinationConfig, Media, NotificationProfile
@@ -25,9 +23,9 @@
 
 
 __all__ = [
+    "are_notifications_turned_on",
     "api_safely_get_medium_object",
     "send_notification",
-    "background_send_notification",
     "find_destinations_for_event",
     "find_destinations_for_many_events",
     "send_notifications_to_users",
@@ -41,6 +39,13 @@
 MEDIA_CLASSES_DICT = {media_class.MEDIA_SLUG: media_class for media_class in MEDIA_CLASSES}
 
 
+def are_notifications_turned_on():
+    if not getattr(settings, "SEND_NOTIFICATIONS", False):
+        LOG.info("Notification: turned off sitewide, not sending any")
+        return False
+    return True
+
+
 def api_safely_get_medium_object(media_slug):
     try:
         obj = MEDIA_CLASSES_DICT[media_slug]
@@ -64,14 +69,6 @@ def send_notification(destinations: Iterable[DestinationConfig], *events: Iterab
                 LOG.warn('Notification: could not send event "%s" to "%s"', event, medium.MEDIA_SLUG)
 
 
-def background_send_notification(destinations: Iterable[DestinationConfig], *events: Event):
-    connections.close_all()
-    LOG.info("Notification: backgrounded: about to send %i events", len(events))
-    p = Process(target=send_notification, args=(destinations, *events))
-    p.start()
-    return p
-
-
 def find_destinations_for_event(event: Event):
     destinations = set()
     incident = event.incident
@@ -97,8 +94,7 @@ def send_notifications_to_users(*events: Iterable[Event], send=send_notification
     if not events:
         LOG.warn("Notification: no events to send, programming error?")
         return
-    if not getattr(settings, "SEND_NOTIFICATIONS", False):
-        LOG.info("Notification: turned off sitewide, not sending any")
+    if not are_notifications_turned_on():
         return
     # TODO: only send one notification per medium per user
     LOG.debug('Fallback filter set to "%s"', getattr(settings, "ARGUS_FALLBACK_FILTER", {}))

src/argus/notificationprofile/models.py

@@ -100,9 +100,10 @@ def save(self, *args, **kwargs):
 class FilterWrapper:
     TRINARY_FILTERS = ("open", "acked", "stateful")
 
-    def __init__(self, filterblob):
+    def __init__(self, filterblob, user=None):
         self.fallback_filter = getattr(settings, "ARGUS_FALLBACK_FILTER", {})
         self.filter = filterblob.copy()
+        self.user = user  # simplifies debugging, set breakpoint for specific user
 
     def _get_tristate(self, tristate):
         fallback_filter = self.fallback_filter.get(tristate, None)
@@ -179,7 +180,8 @@ class Meta:
 
     def __init__(self, *args, **kwargs):
         super().__init__(*args, **kwargs)
-        self.filter_wrapper = FilterWrapper(self.filter)
+        user = getattr(self, "user", None)
+        self.filter_wrapper = FilterWrapper(self.filter, user)
 
     def __str__(self):
         return f"{self.name} [{self.filter}]"

src/argus/site/settings/base.py

@@ -45,6 +45,7 @@
     "drf_spectacular",
     "django_filters",
     "phonenumber_field",
+    "django_q",
 
     # Argus apps
     "argus.auth",
@@ -210,13 +211,14 @@
 
 AUTH_TOKEN_EXPIRES_AFTER_DAYS = 14
 
+# Redis
+_REDIS = urlsplit("//" + get_str_env("ARGUS_REDIS_SERVER", "127.0.0.1:6379"))
 
 # django-channels
 
 ASGI_APPLICATION = "argus.ws.asgi.application"
 
 # fmt: off
-_REDIS = urlsplit("//" + get_str_env("ARGUS_REDIS_SERVER", "127.0.0.1:6379"))
 CHANNEL_LAYERS = {
     "default": {
         "BACKEND": "channels_redis.core.RedisChannelLayer",
@@ -302,3 +304,18 @@
 #
 # SOCIAL_AUTH_DATAPORTEN_FEIDE_KEY = SOCIAL_AUTH_DATAPORTEN_KEY
 # SOCIAL_AUTH_DATAPORTEN_FEIDE_SECRET = SOCIAL_AUTH_DATAPORTEN_SECRET
+
+# Django-Q2
+
+Q_CLUSTER = {
+    "name": "events",
+    "timeout": 60,
+    "time_zone": "UTC",
+    "cpu_affinity": 1,
+    "label": "Django Q2 Queue",
+    "redis": {
+        "host": _REDIS.hostname,
+        "port": _REDIS.port or 6379,
+        "db": 0,
+    },
+}