feat(logs): queue global log purges asynchronously

docs/development.md

@@ -461,6 +461,103 @@ task docker:purge && EDA_SECRET_KEY=insecure task docker:up
 Do **not** confuse the two — see the command docstring in
 `src/aap_eda/core/management/commands/rotate_db_encryption_key.py`.
 
+## Log purge
+
+- **Submit a job:** A superuser sends `POST /api/eda/v1/logs/purge/`. HTTP 202
+  with a `job_id` means dispatcherd accepted the job; deletion may still be
+  running.
+- **Choose a cutoff:** For an age-based purge, supply exactly one of
+  `before_date` (midnight UTC in `YYYY-MM-DD` format) or a positive ISO 8601
+  `retention_duration`. Only rows with `log_timestamp` strictly before the
+  cutoff qualify, including earlier audit markers. Future UTC cutoffs are
+  rejected.
+- **Purge logs globally:** Send `"purge-all-logs": true` without activation
+  selectors. Alone, it deletes logs at every timestamp. Combine it with
+  `before_date` or `retention_duration` to delete only older logs, including
+  unattached logs. Each affected process receives a new audit marker; unattached
+  logs cannot receive one. The deleted count includes earlier markers. If
+  processes keep writing logs, new rows can appear during or after the purge.
+  Without a cutoff or this flag, the API returns HTTP 400 before queueing.
+- **Choose the scope:** With no selectors, qualifying logs from every
+  process are included, including processes without an activation.
+  `activation_ids` and `activation_names` select the union of matching
+  activations. Empty or unknown selectors are rejected before queueing.
+- **Choose audit behavior:** Audit markers are enabled by default. Set API
+  `audit_trail` to `false` to disable them.
+
+For example, this request queues a purge for the named activation:
+
+```http
+POST /api/eda/v1/logs/purge/
+Content-Type: application/json
+
+{"retention_duration": "P30D", "audit_trail": true,
+ "activation_names": ["production-rules"]}
+```
+
+To replace all existing logs with new audit markers:
+
+```http
+POST /api/eda/v1/logs/purge/
+Content-Type: application/json
+
+{"purge-all-logs": true}
+```
+
+To purge logs globally before a cutoff:
+
+```http
+POST /api/eda/v1/logs/purge/
+Content-Type: application/json
+
+{"purge-all-logs": true, "before_date": "2024-01-01"}
+```
+
+**CLI options**
+
+- Use either `--date YYYY-MM-DD` or `--retention-duration P30D`. Calendar
+  months and years in a duration are subtracted in UTC. If the target month
+  lacks the original day, the cutoff uses that month's final day.
+- Use `--purge-all-logs` alone to delete every existing log, or combine it
+  with `--date` or `--retention-duration` to delete only older logs globally.
+  This flag cannot be combined with activation selectors or
+  `--audit-trail=false`. Omitting both a cutoff and this flag raises a command
+  error before deletion.
+- Use `--activation-ids 123 456`, `--activation-names name1 name2`, or both to
+  limit deletion to the selected activations. Omit both to include every
+  activation and processes without an activation.
+- Use `--audit-trail=true` or `--audit-trail=false`. A bare `--audit-trail` is
+  invalid. Default is True.
+- The optional `--internal-job-uuid` accepts a UUID or a custom ID of 1-128
+  ASCII letters, digits, periods, underscores, or hyphens, starting with a
+  letter or digit. Valid UUIDs are normalized. The value appears in audit
+  markers under `Purge Job ID`. The API supplies its task UUID automatically.
+
+**Deletion and audit markers**
+
+- Deletion runs once per process. If a later process fails, worker logs record
+  the count already deleted, and the job reports failure. Audit markers record
+  the count for each affected process when audit is enabled.
+- When audit is enabled, each process that loses log rows gets one marker with
+  its count for this invocation. Markers use `Recorded at` for the time the
+  audit row was written and store that time in `log_created_at`. The API
+  returns this recorded time.
+- Cutoff purges treat earlier markers like other log rows: they are deleted and
+  counted once their `log_timestamp` falls before the cutoff. Recent markers
+  can therefore remain until a later cutoff purge reaches them.
+
+**Check the outcome**
+
+- Inspect default-worker logs by the returned job ID for started, completed,
+  skipped, and failed outcomes, including `deleted_count`:
+
+```shell
+JOB_ID=<job-id-from-response>
+kubectl logs -n "${NAMESPACE:-aap-eda}" \
+  -l app=eda,comp=default-worker --all-containers=true --since=24h \
+  | grep "job_id=${JOB_ID}"
+```
+
 ### Running linters
 
 The project uses `flake8` with a set of plugins, `black`, `isort` and `ruff` (experimental),

pyproject.toml

@@ -68,6 +68,7 @@ pyjwt = { version = ">=2.13.0,<3", extras = ["crypto"] }
 ecdsa = "0.18.*"
 validators = "^0.34.0"
 django-flags = "^5.0.13"
+isodate = "^0.7.2"
 insights-analytics-collector = "^0.3.2"
 distro = "^1.9.0"
 dispatcherd = { version = "v2026.05.27", extras = ["pg_notify"] }

src/aap_eda/api/serializers/__init__.py

@@ -21,6 +21,8 @@
     ActivationReadSerializer,
     ActivationSerializer,
     ActivationUpdateSerializer,
+    LogPurgeRequestSerializer,
+    LogPurgeResponseSerializer,
     PostActivationSerializer,
 )
 from .auth import JWTTokenSerializer, LoginSerializer, RefreshTokenSerializer
@@ -114,6 +116,8 @@
     "ActivationListSerializer",
     "ActivationCreateSerializer",
     "ActivationUpdateSerializer",
+    "LogPurgeRequestSerializer",
+    "LogPurgeResponseSerializer",
     "ActivationReadSerializer",
     "ActivationCopySerializer",
     "ActivationInstanceSerializer",

src/aap_eda/api/serializers/activation.py

@@ -16,6 +16,7 @@
 import secrets
 import uuid
 from dataclasses import dataclass, field
+from datetime import datetime, time, timezone as datetime_timezone
 from typing import Optional
 
 import yaml
@@ -54,7 +55,9 @@
     get_resolved_secrets,
     get_secret_fields,
 )
+from aap_eda.core.utils.delete_log_util import resolve_activation_ids
 from aap_eda.core.utils.k8s_service_name import create_k8s_service_name
+from aap_eda.core.utils.log_retention import retention_cutoff
 from aap_eda.core.utils.rulebook import (
     build_rulebook_with_event_streams,
     build_source_list,
@@ -89,6 +92,139 @@ def _update_event_stream_source(validated_data: dict) -> str:
         raise InvalidEventStreamSource(e) from e
 
 
+class LogPurgeRequestSerializer(serializers.Serializer):
+    """Serializer for the asynchronous log purge endpoint."""
+
+    before_date = serializers.DateField(
+        required=False,
+        help_text="Exclusive UTC midnight cutoff (YYYY-MM-DD).",
+    )
+    retention_duration = serializers.CharField(
+        required=False,
+        help_text="Exclusive cutoff as a positive ISO 8601 duration.",
+    )
+    audit_trail = serializers.BooleanField(
+        required=False,
+        default=True,
+        help_text="Write one audit marker per process that loses logs.",
+    )
+    activation_ids = serializers.ListField(
+        child=serializers.IntegerField(min_value=1),
+        required=False,
+        allow_empty=False,
+        help_text="Limit deletion to these activation IDs.",
+    )
+    activation_names = serializers.ListField(
+        child=serializers.CharField(allow_blank=False, trim_whitespace=False),
+        required=False,
+        allow_empty=False,
+        help_text="Limit deletion to these activation names.",
+    )
+
+    def get_fields(self):
+        fields = super().get_fields()
+        fields["purge-all-logs"] = serializers.BooleanField(
+            source="purge_all_logs",
+            required=False,
+            default=False,
+            help_text=(
+                "Delete all existing process logs globally. Can be combined "
+                "with before_date or retention_duration to limit deletion "
+                "to older logs. Cannot be combined with activation selectors "
+                "or disabled audit."
+            ),
+        )
+        return fields
+
+    @staticmethod
+    def _validate_full_purge(attrs):
+        """Reject selector or audit-off combinations with purge-all-logs."""
+        if "activation_ids" in attrs or "activation_names" in attrs:
+            raise serializers.ValidationError(
+                "purge-all-logs cannot be combined with activation selectors."
+            )
+        if not attrs["audit_trail"]:
+            raise serializers.ValidationError(
+                "purge-all-logs requires audit_trail=true."
+            )
+
+    def _validate_cutoff(self, attrs):
+        """Ensure exactly one cutoff is supplied and normalize it."""
+        before_date = attrs.get("before_date")
+        duration = attrs.get("retention_duration")
+        has_date = before_date is not None
+        has_duration = duration is not None
+        err = "Exactly one of before_date or retention_duration is required."
+
+        if has_date and has_duration:
+            raise serializers.ValidationError(err)
+        if not has_date and not has_duration and not attrs["purge_all_logs"]:
+            raise serializers.ValidationError(err)
+        if has_date:
+            self._validate_before_date(attrs, before_date)
+        elif has_duration:
+            self._validate_duration(duration)
+
+    @staticmethod
+    def _validate_before_date(attrs, before_date):
+        """Reject future dates and normalise before_date to ISO string."""
+        cutoff = datetime.combine(
+            before_date, time.min, tzinfo=datetime_timezone.utc
+        )
+        if cutoff > datetime.now(datetime_timezone.utc):
+            raise serializers.ValidationError(
+                {"before_date": "The UTC cutoff must not be future."}
+            )
+        attrs["before_date"] = before_date.isoformat()
+
+    @staticmethod
+    def _validate_duration(duration):
+        """Validate retention_duration as a positive ISO 8601 duration."""
+        try:
+            retention_cutoff(duration)
+        except ValueError as exc:
+            raise serializers.ValidationError(
+                {"retention_duration": str(exc)}
+            ) from exc
+
+    @staticmethod
+    def _resolve_activations(attrs):
+        """Resolve activation names to IDs."""
+        if "activation_ids" not in attrs and "activation_names" not in attrs:
+            return
+        try:
+            attrs["activation_ids"] = sorted(
+                resolve_activation_ids(
+                    attrs.get("activation_ids"),
+                    attrs.get("activation_names"),
+                )
+            )
+        except ValueError as exc:
+            raise serializers.ValidationError(str(exc)) from exc
+        attrs.pop("activation_names", None)
+
+    def validate(self, attrs):
+        unknown_fields = set(self.initial_data) - set(self.fields)
+        if unknown_fields:
+            raise serializers.ValidationError(
+                dict.fromkeys(sorted(unknown_fields), "Unknown field.")
+            )
+
+        if attrs["purge_all_logs"]:
+            self._validate_full_purge(attrs)
+        self._validate_cutoff(attrs)
+        if attrs["purge_all_logs"]:
+            return attrs
+        self._resolve_activations(attrs)
+        return attrs
+
+
+class LogPurgeResponseSerializer(serializers.Serializer):
+    """Acknowledgement returned after a purge task is queued."""
+
+    job_id = serializers.UUIDField()
+
+
 def _update_k8s_service_name(validated_data: dict) -> str:
     service_name = validated_data.get("k8s_service_name")
     return service_name or create_k8s_service_name(validated_data["name"])

src/aap_eda/api/urls.py

@@ -37,6 +37,7 @@
 router.register("projects", views.ProjectViewSet)
 router.register("rulebooks", views.RulebookViewSet)
 router.register("activations", views.ActivationViewSet)
+router.register("logs", views.LogPurgeViewSet, basename="logs")
 router.register(
     "activation-instances",
     views.ActivationInstanceViewSet,

src/aap_eda/api/views/__init__.py

@@ -12,7 +12,11 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
-from .activation import ActivationInstanceViewSet, ActivationViewSet
+from .activation import (
+    ActivationInstanceViewSet,
+    ActivationViewSet,
+    LogPurgeViewSet,
+)
 from .auth import SessionLoginView, SessionLogoutView, TokenRefreshView
 from .config import ConfigView
 from .credential_input_source import CredentialInputSourceViewSet
@@ -40,6 +44,7 @@
     # activations
     "ActivationViewSet",
     "ActivationInstanceViewSet",
+    "LogPurgeViewSet",
     # user
     "CurrentUserView",
     "CurrentUserAwxTokenViewSet",

src/aap_eda/api/views/activation.py

@@ -31,11 +31,13 @@
 from rest_framework.response import Response
 
 from aap_eda.api import exceptions as api_exc, filters, serializers
+from aap_eda.api.permissions import IsSystemAdmin
 from aap_eda.api.serializers.activation import is_activation_valid
 from aap_eda.core import models
 from aap_eda.core.enums import Action, ActivationStatus, ProcessParentType
 from aap_eda.core.health import check_dispatcherd_workers_health
 from aap_eda.core.utils import logging_utils
+from aap_eda.tasks.management import submit_log_purge
 from aap_eda.tasks.orchestrator import (
     delete_rulebook_process,
     restart_rulebook_process,
@@ -868,3 +870,54 @@ def logs(self, request, id):
             results, many=True
         )
         return self.get_paginated_response(serializer.data)
+
+
+class LogPurgeViewSet(viewsets.GenericViewSet):
+    """Superuser-only log purge scheduling endpoint."""
+
+    permission_classes = (IsSystemAdmin,)
+    serializer_class = serializers.LogPurgeRequestSerializer
+
+    @extend_schema(
+        description=(
+            "Queue deletion of process logs using one UTC cutoff: "
+            "before_date (YYYY-MM-DD) or retention_duration (positive ISO "
+            "8601). Alternatively, purge-all-logs=true deletes logs "
+            "globally. It can be combined with before_date or "
+            "retention_duration to delete only older logs, including prior "
+            "audit markers and logs without a process. It leaves one new "
+            "audit marker per affected process. Logs without a process "
+            "cannot receive a marker. This mode cannot be combined with "
+            "activation selectors or disabled audit. Without activation "
+            "selectors, all processes, including orphaned ones, are included. "
+            "Activation IDs and names limit "
+            "cutoff deletion to their union. The returned job ID "
+            "acknowledges scheduling only."
+        ),
+        request=serializers.LogPurgeRequestSerializer,
+        responses={
+            status.HTTP_202_ACCEPTED: serializers.LogPurgeResponseSerializer,
+            status.HTTP_400_BAD_REQUEST: OpenApiResponse(
+                description="The request contains invalid purge options."
+            ),
+            status.HTTP_403_FORBIDDEN: OpenApiResponse(
+                description="Only superusers can queue a log purge."
+            ),
+            status.HTTP_500_INTERNAL_SERVER_ERROR: OpenApiResponse(
+                description="The purge task could not be queued."
+            ),
+        },
+    )
+    @action(methods=["post"], detail=False, url_path="purge")
+    def purge(self, request):
+        serializer = self.get_serializer(data=request.data)
+        serializer.is_valid(raise_exception=True)
+        try:
+            job_id = submit_log_purge(**serializer.validated_data)
+        except Exception:
+            logger.exception("Unable to queue log purge")
+            return Response(
+                {"detail": "Unable to queue log purge job."},
+                status=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            )
+        return Response({"job_id": job_id}, status=status.HTTP_202_ACCEPTED)