feat(activations): log retention Phase 2 — timestamp filters, DEBUG toggle, purge API, safety valve

src/aap_eda/api/filters/activation.py

@@ -68,7 +68,17 @@ class ActivationInstanceLogFilter(django_filters.FilterSet):
         lookup_expr="icontains",
         label="Filter by activation instance log.",
     )
+    log_timestamp__gt = django_filters.NumberFilter(
+        field_name="log_timestamp",
+        lookup_expr="gt",
+        label="Filter logs with timestamp greater than value.",
+    )
+    log_timestamp__lt = django_filters.NumberFilter(
+        field_name="log_timestamp",
+        lookup_expr="lt",
+        label="Filter logs with timestamp less than value.",
+    )
 
     class Meta:
         model = models.RulebookProcessLog
-        fields = ["log"]
+        fields = ["log", "log_timestamp__gt", "log_timestamp__lt"]

src/aap_eda/api/serializers/__init__.py

@@ -21,6 +21,8 @@
     ActivationReadSerializer,
     ActivationSerializer,
     ActivationUpdateSerializer,
+    LogPurgeRequestSerializer,
+    LogPurgeResponseSerializer,
     PostActivationSerializer,
 )
 from .auth import JWTTokenSerializer, LoginSerializer, RefreshTokenSerializer

src/aap_eda/api/serializers/activation.py

@@ -485,6 +485,7 @@ class Meta:
             "status_message",
             "awx_token_id",
             "log_level",
+            "store_debug_logs",
             "eda_credentials",
             "k8s_service_name",
             "event_streams",
@@ -557,6 +558,7 @@ def to_representation(self, activation):
             "status_message": activation.status_message,
             "awx_token_id": activation.awx_token_id,
             "log_level": activation.log_level,
+            "store_debug_logs": activation.store_debug_logs,
             "eda_credentials": eda_credentials,
             "k8s_service_name": activation.k8s_service_name,
             "event_streams": event_streams,
@@ -595,6 +597,7 @@ class Meta:
             "restart_policy",
             "awx_token_id",
             "log_level",
+            "store_debug_logs",
             "eda_credentials",
             "k8s_service_name",
             "source_mappings",
@@ -750,6 +753,7 @@ def copy(self) -> dict:
             "restart_policy": activation.restart_policy,
             "awx_token_id": activation.awx_token,
             "log_level": activation.log_level,
+            "store_debug_logs": activation.store_debug_logs,
             "eda_credentials": activation.eda_credentials.all(),
             "k8s_service_name": activation.k8s_service_name,
             "source_mappings": activation.source_mappings,
@@ -803,6 +807,7 @@ class Meta:
             "restart_policy",
             "awx_token_id",
             "log_level",
+            "store_debug_logs",
             "eda_credentials",
             "k8s_service_name",
             "source_mappings",
@@ -1157,6 +1162,7 @@ class Meta:
             "awx_token_id",
             "eda_credentials",
             "log_level",
+            "store_debug_logs",
             "k8s_service_name",
             "k8s_pod_service_account_name",
             "k8s_pod_labels",
@@ -1299,6 +1305,7 @@ def to_representation(self, activation):
             "status_message": activation.status_message,
             "awx_token_id": activation.awx_token_id,
             "log_level": activation.log_level,
+            "store_debug_logs": activation.store_debug_logs,
             "eda_credentials": eda_credentials,
             "k8s_service_name": activation.k8s_service_name,
             **_activation_k8s_pod_metadata_payload(activation),
@@ -1816,3 +1823,22 @@ def _validate_persistence_credential(data: dict) -> None:
             f"'{settings.DEFAULT_SYSTEM_RULE_ENGINE_CREDENTIAL_NAME}' "
             "could not be found. Contact your system administrator."
         )
+
+
+class LogPurgeRequestSerializer(serializers.Serializer):
+    """Serializer for log purge request body."""
+
+    before_date = serializers.DateTimeField(
+        required=False,
+        allow_null=True,
+        help_text="Delete logs older than this date. "
+        "If omitted, all logs are deleted.",
+    )
+
+
+class LogPurgeResponseSerializer(serializers.Serializer):
+    """Serializer for log purge response."""
+
+    deleted = serializers.IntegerField(
+        help_text="Number of log records deleted.",
+    )

src/aap_eda/api/urls.py

@@ -56,6 +56,7 @@
 router.register("organizations", views.OrganizationViewSet)
 router.register("teams", views.TeamViewSet)
 router.register("event-streams", views.EventStreamViewSet)
+router.register("logs", views.LogPurgeViewSet, basename="logs")
 router.register(
     "external_event_stream",
     views.ExternalEventStreamViewSet,

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,8 @@
     # activations
     "ActivationViewSet",
     "ActivationInstanceViewSet",
+    # logs
+    "LogPurgeViewSet",
     # user
     "CurrentUserView",
     "CurrentUserAwxTokenViewSet",

src/aap_eda/api/views/activation.py

@@ -37,6 +37,11 @@
 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.core.utils.delete_log_util import (
+    delete_all_logs,
+    delete_logs_for_activation,
+    delete_logs_older_than,
+)
 from aap_eda.tasks.orchestrator import (
     delete_rulebook_process,
     restart_rulebook_process,
@@ -678,6 +683,41 @@ def copy(self, request, pk):
             status=status.HTTP_201_CREATED,
         )
 
+    @extend_schema(
+        request=serializers.LogPurgeRequestSerializer,
+        responses={
+            status.HTTP_200_OK: serializers.LogPurgeResponseSerializer,
+            status.HTTP_404_NOT_FOUND: OpenApiResponse(
+                None, description="Activation not found."
+            ),
+        },
+    )
+    @action(
+        methods=["post"],
+        detail=True,
+        rbac_action=Action.DELETE,
+        url_path="clear-logs",
+    )
+    def clear_logs(self, request, pk):
+        activation = self.get_object()
+        request_serializer = serializers.LogPurgeRequestSerializer(
+            data=request.data,
+        )
+        request_serializer.is_valid(raise_exception=True)
+        before_date = request_serializer.validated_data.get("before_date")
+
+        if before_date:
+            deleted = delete_logs_older_than(
+                before_date, activation_id=activation.id
+            )
+        else:
+            deleted = delete_logs_for_activation(activation.id)
+
+        return Response(
+            serializers.LogPurgeResponseSerializer({"deleted": deleted}).data,
+            status=status.HTTP_200_OK,
+        )
+
     def _sync_project_if_needed(
         self, activation: models.Activation
     ) -> Response | None:
@@ -874,3 +914,37 @@ def logs(self, request, id):
             results, many=True
         )
         return self.get_paginated_response(serializer.data)
+
+
+class LogPurgeViewSet(viewsets.ViewSet):
+    """Global log purge endpoint (admin only)."""
+
+    @extend_schema(
+        request=serializers.LogPurgeRequestSerializer,
+        responses={
+            status.HTTP_200_OK: serializers.LogPurgeResponseSerializer,
+        },
+    )
+    @action(
+        methods=["post"],
+        detail=False,
+        url_path="purge",
+    )
+    def purge(self, request):
+        if not request.user.is_superuser:
+            raise exceptions.PermissionDenied(
+                "Only administrators can purge all logs."
+            )
+
+        request_serializer = serializers.LogPurgeRequestSerializer(
+            data=request.data,
+        )
+        request_serializer.is_valid(raise_exception=True)
+        before_date = request_serializer.validated_data.get("before_date")
+
+        deleted = delete_all_logs(cutoff=before_date)
+
+        return Response(
+            serializers.LogPurgeResponseSerializer({"deleted": deleted}).data,
+            status=status.HTTP_200_OK,
+        )

src/aap_eda/core/migrations/0074_activation_store_debug_logs.py

@@ -0,0 +1,20 @@
+# Generated by Django 5.2.9 on 2026-08-11 02:17
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("core", "0073_activation_k8s_pod_tolerations"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="activation",
+            name="store_debug_logs",
+            field=models.BooleanField(
+                default=False,
+                help_text="When false (default), DEBUG-level log lines are sent to container stdout but not stored in the database.",
+            ),
+        ),
+    ]

src/aap_eda/core/models/activation.py

@@ -143,6 +143,13 @@ class Meta:
         choices=RulebookProcessLogLevel.choices(),
         default=get_default_log_level,
     )
+    store_debug_logs = models.BooleanField(
+        default=False,
+        help_text=(
+            "When false (default), DEBUG-level log lines are sent to "
+            "container stdout but not stored in the database."
+        ),
+    )
     eda_credentials = models.ManyToManyField(
         "EdaCredential", related_name="activations", default=None
     )

src/aap_eda/core/utils/delete_log_util.py

@@ -12,24 +12,88 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 
+import logging
 from datetime import datetime
 
 from django.db.models import Q
 from django.utils import timezone
 
 from aap_eda.core import models
 
+logger = logging.getLogger(__name__)
 
-def delete_logs_older_than(cutoff: datetime) -> int:
+BATCH_SIZE = 10_000
+
+
+def delete_logs_older_than(
+    cutoff: datetime,
+    activation_id: int | None = None,
+) -> int:
     """Delete all RulebookProcessLog records older than the cutoff.
 
+    If activation_id is provided, only delete logs for instances
+    belonging to that activation.
+
     Returns the number of records deleted.
     """
     cutoff_ts = int(cutoff.timestamp())
-    deleted, _ = models.RulebookProcessLog.objects.filter(
+    qs = models.RulebookProcessLog.objects.filter(
         log_timestamp__lt=cutoff_ts,
-    ).delete()
-    return deleted
+    )
+    if activation_id is not None:
+        instance_ids = models.RulebookProcess.objects.filter(
+            activation_id=activation_id,
+        ).values_list("id", flat=True)
+        qs = qs.filter(activation_instance_id__in=instance_ids)
+    return _batched_delete(qs)
+
+
+def delete_logs_for_activation(activation_id: int) -> int:
+    """Delete all logs for a given activation's instances.
+
+    Returns the number of records deleted.
+    """
+    instance_ids = models.RulebookProcess.objects.filter(
+        activation_id=activation_id,
+    ).values_list("id", flat=True)
+    qs = models.RulebookProcessLog.objects.filter(
+        activation_instance_id__in=instance_ids,
+    )
+    return _batched_delete(qs)
+
+
+def delete_all_logs(cutoff: datetime | None = None) -> int:
+    """Delete all RulebookProcessLog records.
+
+    If cutoff is provided, only delete logs older than the cutoff.
+
+    Returns the number of records deleted.
+    """
+    qs = models.RulebookProcessLog.objects.all()
+    if cutoff is not None:
+        cutoff_ts = int(cutoff.timestamp())
+        qs = qs.filter(log_timestamp__lt=cutoff_ts)
+    return _batched_delete(qs)
+
+
+def _batched_delete(queryset) -> int:
+    """Delete queryset in batches to avoid long-running queries."""
+    total_deleted = 0
+    upper_id = queryset.order_by("-id").values_list("id", flat=True).first()
+    if upper_id is None:
+        return total_deleted
+    queryset = queryset.filter(id__lte=upper_id)
+
+    while True:
+        batch_ids = list(queryset.values_list("id", flat=True)[:BATCH_SIZE])
+        if not batch_ids:
+            break
+        deleted, _ = models.RulebookProcessLog.objects.filter(
+            id__in=batch_ids,
+        ).delete()
+        total_deleted += deleted
+        logger.info("Purged %d log records (batch)", deleted)
+    return total_deleted
 
 
 def create_audit_trail(

src/aap_eda/services/activation/activation_manager.py

@@ -14,6 +14,7 @@
 """Module for Activation Manager."""
 
 import contextlib
+import functools
 import logging
 import typing as tp
 from datetime import timedelta
@@ -76,7 +77,10 @@ def __init__(
                 db_instance.id, self.db_instance_type
             )
 
-        self.container_logger_class = container_logger_class
+        self.container_logger_class = functools.partial(
+            container_logger_class,
+            store_debug_logs=getattr(db_instance, "store_debug_logs", False),
+        )
 
     def _update_started_time(self) -> None:
         """Update latest instance's started_at to now."""