[change] Send VPN sync errors/recovery as generic_message notifications #1049

stktyagi · web-flow · commit 012a0942bf4d · 2026-02-07T14:47:25.000-03:00
Aimed at zerotier and wireguard. Made existing code reusable. Closes #1049
diff --git a/openwisp_controller/config/tasks.py b/openwisp_controller/config/tasks.py
@@ -9,6 +9,8 @@
 
 from openwisp_utils.tasks import OpenwispCeleryTask
 
+from .utils import handle_error_notification, handle_recovery_notification
+
 logger = logging.getLogger(__name__)
 
 
@@ -130,18 +132,35 @@ def trigger_vpn_server_endpoint(endpoint, auth_token, vpn_id):
 
     # Cache the configuration here makes downloading the configuration faster.
     vpn.get_cached_configuration()
-    response = requests.post(
-        endpoint,
-        params={"key": auth_token},
-        verify=False if getattr(settings, "DEBUG") else True,
-    )
-    if response.status_code == 200:
-        logger.info(f"Triggered update webhook of VPN Server UUID: {vpn_id}")
+    task_key = f"vpn_update_task:{vpn_id}"
+    try:
+        response = requests.post(
+            endpoint,
+            params={"key": auth_token},
+            verify=not settings.DEBUG,
+            timeout=30,
+        )
+        response.raise_for_status()
+    except requests.RequestException as e:
+        logger.warning(
+            f"Failed to update VPN Server configuration. "
+            f"Error: {str(e)}, "
+            f"VPN Server UUID: {vpn_id}"
+        )
+        handle_error_notification(
+            task_key,
+            sleep_time=5,
+            exception=e,
+            instance=vpn,
+            action="update",
+        )
     else:
-        logger.error(
-            "Failed to update VPN Server configuration. "
-            f"Response status code: {response.status_code}, "
-            f"VPN Server UUID: {vpn_id}",
+        logger.info(f"Triggered update webhook of VPN Server UUID: {vpn_id}")
+        handle_recovery_notification(
+            task_key,
+            sleep_time=5,
+            instance=vpn,
+            action="update",
         )
 
 
diff --git a/openwisp_controller/config/tasks_zerotier.py b/openwisp_controller/config/tasks_zerotier.py
@@ -1,19 +1,16 @@
 import logging
 from http import HTTPStatus
-from time import sleep
 
 from celery import shared_task
-from django.core.cache import cache
 from django.core.exceptions import ObjectDoesNotExist
-from django.utils.translation import gettext as _
-from openwisp_notifications.signals import notify
 from requests.exceptions import RequestException
 from swapper import load_model
 
 from openwisp_controller.config.api.zerotier_service import ZerotierService
 from openwisp_utils.tasks import OpenwispCeleryTask
 
 from .settings import API_TASK_RETRY_OPTIONS
+from .utils import handle_error_notification, handle_recovery_notification
 
 logger = logging.getLogger(__name__)
 
@@ -27,48 +24,6 @@ class OpenwispApiTask(OpenwispCeleryTask):
         HTTPStatus.GATEWAY_TIMEOUT,  # 504
     ]
 
-    def _send_api_task_notification(self, type, **kwargs):
-        vpn = kwargs.get("instance")
-        action = kwargs.get("action").replace("_", " ")
-        status_code = kwargs.get("status_code")
-        # Adding some delay here to prevent overlapping
-        # of the django success message container
-        # with the ow-notification container
-        # https://github.com/openwisp/openwisp-notifications/issues/264
-        sleep(2)
-        message_map = {
-            "error": {
-                "verb": _("encountered an unrecoverable error"),
-                "message": _(
-                    "Unable to perform {action} operation on the "
-                    "{target} VPN server due to an "
-                    "unrecoverable error "
-                    "(status code: {status_code})"
-                ),
-                "level": "error",
-            },
-            "recovery": {
-                "verb": _("has been completed successfully"),
-                "message": _("The {action} operation on {target} {verb}."),
-                "level": "info",
-            },
-        }
-        meta = message_map[type]
-        notify.send(
-            type="generic_message",
-            sender=vpn,
-            target=vpn,
-            action=action,
-            verb=meta["verb"],
-            message=meta["message"].format(
-                action=action,
-                target=str(vpn),
-                status_code=status_code,
-                verb=meta["verb"],
-            ),
-            level=meta["level"],
-        )
-
     def handle_api_call(self, fn, *args, send_notification=True, **kwargs):
         """
         This method handles API calls and their responses
@@ -105,13 +60,10 @@ def handle_api_call(self, fn, *args, send_notification=True, **kwargs):
             response.raise_for_status()
             logger.info(info_msg)
             if send_notification:
-                task_result = cache.get(task_key)
-                if task_result == "error":
-                    self._send_api_task_notification("recovery", **kwargs)
-                    cache.set(task_key, "success", None)
+                handle_recovery_notification(task_key, **kwargs)
         except RequestException as e:
             if response.status_code in self._RECOVERABLE_API_CODES:
-                retry_logger = logger.warn
+                retry_logger = logger.warning
                 # When retry limit is reached, use error logging
                 if self.request.retries == self.max_retries:
                     retry_logger = logger.error
@@ -122,12 +74,7 @@ def handle_api_call(self, fn, *args, send_notification=True, **kwargs):
                 raise e
             logger.error(f"{err_msg}, Error: {e}")
             if send_notification:
-                task_result = cache.get(task_key)
-                if task_result in (None, "success"):
-                    cache.set(task_key, "error", None)
-                    self._send_api_task_notification(
-                        "error", status_code=response.status_code, **kwargs
-                    )
+                handle_error_notification(task_key, exception=e, **kwargs)
         return (response, updated_config) if updated_config else response
 
 
@@ -150,6 +97,8 @@ def trigger_zerotier_server_update(self, config, vpn_id):
         network_id,
         instance=vpn,
         action="update",
+        # notification kwargs
+        sleep_time=5,
         info=(
             f"Successfully updated the configuration of "
             f"ZeroTier VPN Server with UUID: {vpn_id}"
@@ -189,6 +138,8 @@ def trigger_zerotier_server_update_member(self, vpn_id, ip=None, node_id=None):
         member_ip,
         instance=vpn,
         action="update_member",
+        # notification kwargs
+        sleep_time=5,
         info=(
             f"Successfully updated ZeroTier network member: {node_id}, "
             f"ZeroTier network: {network_id}, "
@@ -216,7 +167,7 @@ def trigger_zerotier_server_remove_member(self, node_id=None, **vpn_kwargs):
     network_id = vpn_kwargs.get("network_id")
     try:
         vpn = Vpn.objects.get(pk=vpn_id)
-        notification_kwargs = dict(instance=vpn, action="remove_member")
+        notification_kwargs = dict(instance=vpn, action="remove_member", sleep_time=5)
     # When a ZeroTier VPN server is deleted
     # and this is followed by the deletion of ZeroTier VPN clients
     # we won't have access to the VPN server instance. Therefore, we should
@@ -264,6 +215,7 @@ def trigger_zerotier_server_join(self, vpn_id):
         network_id,
         instance=vpn,
         action="network_join",
+        sleep_time=5,
         info=(
             f"Successfully joined the ZeroTier network: {network_id}, "
             f"ZeroTier VPN Server UUID: {vpn_id}"
diff --git a/openwisp_controller/config/tests/test_notifications.py b/openwisp_controller/config/tests/test_notifications.py
@@ -31,7 +31,9 @@ class TestNotifications(
     app_label = "config"
     _ZT_SERVICE_REQUESTS = "openwisp_controller.config.api.zerotier_service.requests"
     _ZT_API_TASKS_INFO_LOGGER = "openwisp_controller.config.tasks_zerotier.logger.info"
-    _ZT_API_TASKS_WARN_LOGGER = "openwisp_controller.config.tasks_zerotier.logger.warn"
+    _ZT_API_TASKS_WARN_LOGGER = (
+        "openwisp_controller.config.tasks_zerotier.logger.warning"
+    )
     _ZT_API_TASKS_ERR_LOGGER = "openwisp_controller.config.tasks_zerotier.logger.error"
     # As the locmem cache does not support the redis backend cache.keys() method
     _ZT_API_TASKS_LOCMEM_CACHE_KEYS = f"{settings.CACHES['default']['BACKEND']}.keys"
diff --git a/openwisp_controller/config/tests/test_vpn.py b/openwisp_controller/config/tests/test_vpn.py
@@ -3,12 +3,12 @@
 from subprocess import CalledProcessError, TimeoutExpired
 from unittest import mock
 
+import requests
 from celery.exceptions import Retry, SoftTimeLimitExceeded
 from django.conf import settings
 from django.core.exceptions import ValidationError
 from django.db.models.signals import post_save
 from django.db.utils import IntegrityError
-from django.http.response import HttpResponse, HttpResponseNotFound
 from django.test import TestCase, TransactionTestCase
 from requests.exceptions import ConnectionError, RequestException, Timeout
 from swapper import load_model
@@ -728,9 +728,12 @@ def test_trigger_vpn_server_endpoint_invalid_vpn_id(self):
 
 
 class TestWireguardTransaction(BaseTestVpn, TestWireguardVpnMixin, TransactionTestCase):
+    mock_response = mock.Mock(spec=requests.Response)
+    mock_response.status_code = 200
+    mock_response.raise_for_status = mock.Mock()
+
     @mock.patch(
-        "openwisp_controller.config.tasks.requests.post",
-        return_value=HttpResponse(status=200),
+        "openwisp_controller.config.tasks.requests.post", return_value=mock_response
     )
     def test_auto_peer_configuration(self, *args):
         self.assertEqual(IpAddress.objects.count(), 0)
@@ -824,6 +827,9 @@ def test_update_vpn_server_configuration(self):
                     f"Cannot update configuration of {vpn.name} VPN server, "
                     "webhook endpoint and authentication token are empty."
                 )
+        success_response = mock.Mock(spec=requests.Response)
+        success_response.status_code = 200
+        success_response.raise_for_status = mock.Mock()
 
         with self.subTest("Webhook endpoint and authentication endpoint is present"):
             vpn.webhook_endpoint = "https://example.com"
@@ -834,24 +840,29 @@ def test_update_vpn_server_configuration(self):
             with mock.patch(
                 "openwisp_controller.config.tasks.logger.info"
             ) as mocked_logger, mock.patch(
-                "requests.post", return_value=HttpResponse()
+                "requests.post", return_value=success_response
             ):
                 post_save.send(
                     instance=vpn_client, sender=vpn_client._meta.model, created=False
                 )
                 mocked_logger.assert_called_once_with(
                     f"Triggered update webhook of VPN Server UUID: {vpn.pk}"
                 )
+            fail_response = mock.Mock(spec=requests.Response)
+            fail_response.status_code = 404
+            fail_response.raise_for_status.side_effect = requests.exceptions.HTTPError(
+                "Not Found"
+            )
 
-            with mock.patch("logging.Logger.error") as mocked_logger, mock.patch(
-                "requests.post", return_value=HttpResponseNotFound()
+            with mock.patch("logging.Logger.warning") as mocked_logger, mock.patch(
+                "requests.post", return_value=fail_response
             ):
                 post_save.send(
                     instance=vpn_client, sender=vpn_client._meta.model, created=False
                 )
                 mocked_logger.assert_called_once_with(
                     "Failed to update VPN Server configuration. "
-                    f"Response status code: 404, VPN Server UUID: {vpn.pk}"
+                    f"Error: Not Found, VPN Server UUID: {vpn.pk}"
                 )
 
     def test_vpn_peers_changed(self):
@@ -985,9 +996,12 @@ def test_auto_client(self):
 class TestVxlanTransaction(
     BaseTestVpn, TestVxlanWireguardVpnMixin, TransactionTestCase
 ):
+    mock_response = mock.Mock(spec=requests.Response)
+    mock_response.status_code = 200
+    mock_response.raise_for_status = mock.Mock()
+
     @mock.patch(
-        "openwisp_controller.config.tasks.requests.post",
-        return_value=HttpResponse(status=200),
+        "openwisp_controller.config.tasks.requests.post", return_value=mock_response
     )
     def test_auto_peer_configuration(self, *args):
         self.assertEqual(IpAddress.objects.count(), 0)
@@ -1460,7 +1474,9 @@ class TestZeroTierTransaction(
 ):
     _ZT_SERVICE_REQUESTS = "openwisp_controller.config.api.zerotier_service.requests"
     _ZT_API_TASKS_INFO_LOGGER = "openwisp_controller.config.tasks_zerotier.logger.info"
-    _ZT_API_TASKS_WARN_LOGGER = "openwisp_controller.config.tasks_zerotier.logger.warn"
+    _ZT_API_TASKS_WARN_LOGGER = (
+        "openwisp_controller.config.tasks_zerotier.logger.warning"
+    )
     _ZT_API_TASKS_ERR_LOGGER = "openwisp_controller.config.tasks_zerotier.logger.error"
     # As the locmem cache does not support the redis backend cache.keys() method
     _ZT_API_TASKS_LOCMEM_CACHE_KEYS = f"{settings.CACHES['default']['BACKEND']}.keys"
diff --git a/openwisp_controller/config/utils.py b/openwisp_controller/config/utils.py
@@ -1,10 +1,14 @@
 import logging
+import time
 
+from django.core.cache import cache
 from django.core.exceptions import ValidationError
 from django.db.models import Q
 from django.http import Http404, HttpResponse
 from django.shortcuts import get_object_or_404 as base_get_object_or_404
 from django.urls import path, re_path
+from django.utils.translation import gettext_lazy as _
+from openwisp_notifications.signals import notify
 from openwisp_notifications.utils import _get_object_link
 
 logger = logging.getLogger(__name__)
@@ -206,3 +210,62 @@ def get_default_templates_queryset(
 def get_config_error_notification_target_url(obj, field, absolute_url=True):
     url = _get_object_link(obj._related_object(field), absolute_url)
     return f"{url}#config-group"
+
+
+def send_api_task_notification(type, sleep_time=False, **kwargs):
+    """
+    The sleep_time argument is needed to avoid triggering the toast
+    notification in the admin while the page is reloading.
+    """
+    if sleep_time:
+        time.sleep(sleep_time)
+    vpn = kwargs.get("instance")
+    action = kwargs.get("action", "").replace("_", " ")
+    exception = kwargs.get("exception")
+    message_map = {
+        "error": {
+            "verb": _("encountered an unrecoverable error"),
+            "message": _(
+                "Unable to perform {action} operation on the "
+                "{target} VPN server due to an "
+                "unrecoverable error "
+                "({error_type})"
+            ),
+            "level": "error",
+        },
+        "success": {
+            "verb": _("has been completed successfully"),
+            "message": _("The {action} operation on {target} {verb}."),
+            "level": "info",
+        },
+    }
+    meta = message_map[type]
+    notify.send(
+        sender=vpn,
+        target=vpn,
+        type="generic_message",
+        action_object=vpn,
+        verb=meta["verb"],
+        message=meta["message"].format(
+            action=action,
+            target=str(vpn),
+            error_type=exception.__class__.__name__ if exception else "",
+            verb=meta["verb"],
+        ),
+        description=str(exception) if exception else "",
+        level=meta["level"],
+    )
+
+
+def handle_recovery_notification(task_key, sleep_time=False, **kwargs):
+    task_result = cache.get(task_key)
+    if task_result == "error":
+        send_api_task_notification("success", sleep_time=sleep_time, **kwargs)
+    cache.set(task_key, "success", timeout=None)
+
+
+def handle_error_notification(task_key, sleep_time=False, **kwargs):
+    cached_value = cache.get(task_key)
+    if cached_value != "error":
+        cache.set(task_key, "error", timeout=None)
+        send_api_task_notification("error", sleep_time=sleep_time, **kwargs)
