[fix] Log metrics for bulk-closed RADIUS sessions #734

The RADIUS monitoring integration previously relied only on
Django's ``post_save`` signal to write traffic snapshots.

However, sessions closed via ``QuerySet.update()``
(``Accounting-On`` handling) and ``bulk_update()``
(``close_previous_radius_accounting_sessions``) bypassed this
signal, causing traffic to be missing from RADIUS traffic
charts.

Sessions closed via ``QuerySet.update()`` and ``bulk_update()``
now emit the ``radius_accounting_closed`` signal, ensuring
that traffic metrics are correctly snapshotted by the
RADIUS monitoring integration.

Closes #734

Author: nemesifier

docs/developer/utils.rst

@@ -45,7 +45,10 @@ Editing an already closed session does not emit this signal again.
 The signal is emitted after the database transaction is committed.
 
 Integrations which need to react to closed accounting sessions should
-listen to this signal.
+listen to this signal. For example, the monitoring integration uses this
+signal to write RADIUS traffic snapshots for sessions closed by regular
+``post_save`` paths, ``Accounting-On`` packets and automatic stale-session
+cleanup.
 
 When closing multiple sessions with ``bulk_update()``, use
 ``RadiusAccounting.emit_radius_accounting_closed()`` after the database

openwisp_radius/base/models.py

@@ -627,13 +627,39 @@ def _close_stale_sessions_on_nas_boot(cls, called_station_id):
         """
         if not called_station_id:
             return 0
-        stale_sessions = cls.objects.filter(
-            called_station_id=called_station_id,
-            stop_time__isnull=True,
-        )
-        closed_count = stale_sessions.update(
-            stop_time=now(), terminate_cause="NAS-Reboot"
-        )
+        stop_time = now()
+        closed_count = 0
+        batch_size = 1000
+        has_more_sessions = True
+        while has_more_sessions:
+            with transaction.atomic():
+                closed_sessions = list(
+                    cls.objects.select_for_update()
+                    .filter(
+                        called_station_id=called_station_id,
+                        stop_time__isnull=True,
+                    )
+                    .only(
+                        "unique_id",
+                        "username",
+                        "organization_id",
+                        "input_octets",
+                        "output_octets",
+                        "calling_station_id",
+                        "called_station_id",
+                        "stop_time",
+                    )[:batch_size]
+                )
+                has_more_sessions = len(closed_sessions) == batch_size
+                if not closed_sessions:
+                    continue
+                for session in closed_sessions:
+                    session.stop_time = stop_time
+                    session.terminate_cause = "NAS-Reboot"
+                closed_count += cls.objects.bulk_update(
+                    closed_sessions, fields=["stop_time", "terminate_cause"]
+                )
+                cls.emit_radius_accounting_closed(closed_sessions)
         return closed_count
 
 

openwisp_radius/integrations/monitoring/management/__init__.py

@@ -0,0 +1,122 @@
+from django.core.management.base import BaseCommand, CommandError
+from django.utils import timezone
+from django.utils.dateparse import parse_datetime
+from openwisp_monitoring.db import timeseries_db
+from swapper import load_model
+
+from openwisp_radius.integrations.monitoring import tasks
+from openwisp_radius.integrations.monitoring.utils import sha1_hash
+
+RadiusAccounting = load_model("openwisp_radius", "RadiusAccounting")
+
+
+class Command(BaseCommand):
+    """
+    Rebuild RADIUS accounting metrics missed by the NAS-Reboot bulk update path.
+
+    See https://github.com/openwisp/openwisp-radius/issues/734.
+    TODO: remove in version 1.4.
+    """
+
+    help = "Private command to rebuild monitoring metrics for closed RADIUS sessions."
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--commit",
+            action="store_true",
+            help="Write metrics. Without this flag the command only reports the count.",
+        )
+        parser.add_argument(
+            "--start",
+            help=(
+                "Only process sessions with stop_time greater than or equal "
+                "to this date."
+            ),
+        )
+        parser.add_argument(
+            "--end",
+            help=(
+                "Only process sessions with stop_time lower than or equal "
+                "to this date."
+            ),
+        )
+        parser.add_argument(
+            "--chunk-size",
+            type=int,
+            default=1000,
+            help="Number of sessions fetched per database batch.",
+        )
+
+    def handle(self, *args, **options):
+        queryset = RadiusAccounting.objects.filter(
+            stop_time__isnull=False,
+            terminate_cause="NAS-Reboot",
+        )
+        if options["start"]:
+            queryset = queryset.filter(
+                stop_time__gte=self._parse_datetime(options["start"], "start")
+            )
+        if options["end"]:
+            queryset = queryset.filter(
+                stop_time__lte=self._parse_datetime(options["end"], "end")
+            )
+        queryset = queryset.order_by("stop_time", "unique_id").only(
+            "unique_id",
+            "username",
+            "organization_id",
+            "input_octets",
+            "output_octets",
+            "calling_station_id",
+            "called_station_id",
+            "stop_time",
+            "terminate_cause",
+        )
+        count = queryset.count()
+        if not options["commit"]:
+            self.stdout.write(f"Dry run: {count} closed sessions would be processed.")
+            return
+        processed = 0
+        chunk_size = options["chunk_size"]
+        self.stdout.write(f"Starting to rebuild {count} accounting metrics.")
+        for session in queryset.iterator(chunk_size=chunk_size):
+            self._delete_radius_accounting_metric(session)
+            tasks.post_save_radiusaccounting(
+                username=session.username,
+                organization_id=str(session.organization_id),
+                input_octets=session.input_octets,
+                output_octets=session.output_octets,
+                calling_station_id=session.calling_station_id,
+                called_station_id=session.called_station_id,
+                time=session.stop_time,
+            )
+            processed += 1
+            if processed % chunk_size == 0 or processed == count:
+                self.stdout.write(
+                    f"Processed {processed} of {count} accounting metrics."
+                )
+
+    def _parse_datetime(self, value, option):
+        parsed = parse_datetime(value)
+        if parsed is None:
+            raise CommandError(f"Invalid --{option} datetime: {value}")
+        if timezone.is_naive(parsed):
+            parsed = timezone.make_aware(parsed)
+        return parsed
+
+    def _delete_radius_accounting_metric(self, session):
+        tags = {
+            "organization_id": str(session.organization_id),
+            "calling_station_id": sha1_hash(session.calling_station_id),
+            "called_station_id": session.called_station_id,
+        }
+        where = " AND ".join(
+            f"\"{key}\" = '{self._escape_tag_value(value)}'"
+            for key, value in tags.items()
+        )
+        timeseries_db.query(
+            "DELETE FROM radius_acc "
+            f"WHERE time = '{session.stop_time.isoformat()}' AND {where}"
+        )
+
+    def _escape_tag_value(self, value):
+        return str(value).replace("'", r"\'")

openwisp_radius/integrations/monitoring/management/commands/__init__.py

@@ -0,0 +1,150 @@
+from io import StringIO
+from unittest.mock import patch
+
+from django.contrib.contenttypes.models import ContentType
+from django.core.management import call_command
+from django.test import tag
+from django.utils import timezone
+from swapper import load_model
+
+from openwisp_radius.integrations.monitoring.tests.mixins import (
+    CreateDeviceMonitoringMixin,
+)
+from openwisp_radius.integrations.monitoring.utils import sha1_hash
+from openwisp_radius.tests import _RADACCT
+from openwisp_radius.tests.mixins import BaseTransactionTestCase
+
+RegisteredUser = load_model("openwisp_radius", "RegisteredUser")
+RadiusAccounting = load_model("openwisp_radius", "RadiusAccounting")
+
+
+@tag("radius_monitoring", "rebuild_radius_accounting_metrics")
+class TestRebuildRadiusAccountingMetrics(
+    CreateDeviceMonitoringMixin, BaseTransactionTestCase
+):
+    def _create_registered_user(self, **kwargs):
+        options = {
+            "is_verified": False,
+            "method": "mobile_phone",
+            "organization": self.default_org,
+        }
+        options.update(**kwargs)
+        if "user" not in options:
+            options["user"] = self._create_user()
+        reg_user = RegisteredUser(**options)
+        reg_user.full_clean()
+        reg_user.save()
+        return reg_user
+
+    def _create_closed_accounting_without_metric(self, **kwargs):
+        options = _RADACCT.copy()
+        options.update(
+            {
+                "unique_id": "closed-without-metric",
+                "calling_station_id": "00:00:00:00:00:00",
+                "input_octets": 8000000000,
+                "output_octets": 9000000000,
+            }
+        )
+        options.update(kwargs)
+        stop_time = options.pop("stop_time", timezone.now())
+        terminate_cause = options.pop("terminate_cause", "NAS-Reboot")
+        session = self._create_radius_accounting(**options)
+        RadiusAccounting.objects.filter(pk=session.pk).update(
+            stop_time=stop_time,
+            terminate_cause=terminate_cause,
+        )
+        session.refresh_from_db()
+        return session
+
+    def test_rebuild_radius_accounting_metrics_dry_run(self):
+        user = self._create_user()
+        device = self._create_device()
+        self._create_registered_user(user=user)
+        self._create_closed_accounting_without_metric(
+            username=user.username,
+            called_station_id=device.mac_address.replace("-", ":").upper(),
+        )
+        out = StringIO()
+        call_command("rebuild_radius_accounting_metrics", stdout=out)
+        self.assertIn("Dry run: 1 closed sessions would be processed.", out.getvalue())
+        self.assertEqual(
+            self.metric_model.objects.filter(configuration="radius_acc").count(), 0
+        )
+
+    @patch(
+        "openwisp_radius.integrations.monitoring.management.commands."
+        "rebuild_radius_accounting_metrics.timeseries_db.query"
+    )
+    @patch("logging.Logger.warning")
+    def test_rebuild_radius_accounting_metrics_commit(
+        self, mocked_warning, mocked_query
+    ):
+        user = self._create_user()
+        reg_user = self._create_registered_user(user=user)
+        device = self._create_device()
+        device_loc = self._create_device_location(
+            content_object=device,
+            location=self._create_location(organization=device.organization),
+        )
+        session = self._create_closed_accounting_without_metric(
+            username=user.username,
+            called_station_id=device.mac_address.replace("-", ":").upper(),
+        )
+        out = StringIO()
+        call_command("rebuild_radius_accounting_metrics", commit=True, stdout=out)
+        output = out.getvalue()
+        self.assertIn("Starting to rebuild 1 accounting metrics.", output)
+        self.assertIn("Processed 1 of 1 accounting metrics.", output)
+        delete_queries = [
+            call.args[0]
+            for call in mocked_query.call_args_list
+            if call.args[0].startswith("DELETE FROM radius_acc")
+        ]
+        self.assertEqual(len(delete_queries), 1)
+        self.assertEqual(
+            self.metric_model.objects.filter(
+                configuration="radius_acc",
+                name="RADIUS Accounting",
+                key="radius_acc",
+                object_id=str(device.id),
+                content_type=ContentType.objects.get_for_model(self.device_model),
+                extra_tags={
+                    "called_station_id": device.mac_address,
+                    "calling_station_id": sha1_hash(session.calling_station_id),
+                    "location_id": str(device_loc.location.id),
+                    "method": reg_user.method,
+                    "organization_id": str(self.default_org.id),
+                },
+            ).count(),
+            1,
+        )
+
+    @patch("logging.Logger.warning")
+    def test_rebuild_radius_accounting_metrics_nas_reboot_filter(self, *args):
+        user = self._create_user()
+        device = self._create_device()
+        self._create_registered_user(user=user)
+        self._create_closed_accounting_without_metric(
+            unique_id="matching-session",
+            username=user.username,
+            called_station_id=device.mac_address.replace("-", ":").upper(),
+        )
+        self._create_closed_accounting_without_metric(
+            unique_id="ignored-session",
+            username=user.username,
+            called_station_id=device.mac_address.replace("-", ":").upper(),
+            terminate_cause="Session-Timeout",
+        )
+        out = StringIO()
+        call_command(
+            "rebuild_radius_accounting_metrics",
+            commit=True,
+            stdout=out,
+        )
+        output = out.getvalue()
+        self.assertIn("Starting to rebuild 1 accounting metrics.", output)
+        self.assertIn("Processed 1 of 1 accounting metrics.", output)
+        metric = self.metric_model.objects.get(configuration="radius_acc")
+        points = metric.chart_set.get(configuration="radius_traffic").read()
+        self.assertEqual(points["summary"], {"upload": 9, "download": 8})